/*-------------------------------------------------------------------------
 *
 * twophase.c
 *        Two-phase commit support functions.
 *
 * Portions Copyright (c) 1996-2017, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 * Portions Copyright (c) 2010-2012 Postgres-XC Development Group
 *
 * IDENTIFICATION
 *        src/backend/access/transam/twophase.c
 *
 * NOTES
 *        Each global transaction is associated with a global transaction
 *        identifier (GID). The client assigns a GID to a postgres
 *        transaction with the PREPARE TRANSACTION command.
 *
 *        We keep all active global transactions in a shared memory array.
 *        When the PREPARE TRANSACTION command is issued, the GID is
 *        reserved for the transaction in the array. This is done before
 *        a WAL entry is made, because the reservation checks for duplicate
 *        GIDs and aborts the transaction if there already is a global
 *        transaction in prepared state with the same GID.
 *
 *        A global transaction (gxact) also has dummy PGXACT and PGPROC; this is
 *        what keeps the XID considered running by TransactionIdIsInProgress.
 *        It is also convenient as a PGPROC to hook the gxact's locks to.
 *
 *        Information to recover prepared transactions in case of crash is
 *        now stored in WAL for the common case. In some cases there will be
 *        an extended period between preparing a GXACT and commit/abort, in
 *        which case we need to separately record prepared transaction data
 *        in permanent storage. This includes locking information, pending
 *        notifications etc. All that state information is written to the
 *        per-transaction state file in the pg_twophase directory.
 *        All prepared transactions will be written prior to shutdown.
 *
 *        Life track of state data is following:
 *
 *        * On PREPARE TRANSACTION backend writes state data only to the WAL and
 *          stores pointer to the start of the WAL record in
 *          gxact->prepare_start_lsn.
 *        * If COMMIT occurs before checkpoint then backend reads data from WAL
 *          using prepare_start_lsn.
 *        * On checkpoint state data copied to files in pg_twophase directory and
 *          fsynced
 *        * If COMMIT happens after checkpoint then backend reads state data from
 *          files
 *
 *        During replay and replication, TwoPhaseState also holds information
 *        about active prepared transactions that haven't been moved to disk yet.
 *
 *        Replay of twophase records happens by the following rules:
 *
 *        * At the beginning of recovery, pg_twophase is scanned once, filling
 *          TwoPhaseState with entries marked with gxact->inredo and
 *          gxact->ondisk.  Two-phase file data older than the XID horizon of
 *          the redo position are discarded.
 *        * On PREPARE redo, the transaction is added to TwoPhaseState->prepXacts.
 *          gxact->inredo is set to true for such entries.
 *        * On Checkpoint we iterate through TwoPhaseState->prepXacts entries
 *          that have gxact->inredo set and are behind the redo_horizon. We
 *          save them to disk and then switch gxact->ondisk to true.
 *        * On COMMIT/ABORT we delete the entry from TwoPhaseState->prepXacts.
 *          If gxact->ondisk is true, the corresponding entry from the disk
 *          is additionally deleted.
 *        * RecoverPreparedTransactions(), StandbyRecoverPreparedTransactions()
 *          and PrescanPreparedTransactions() have been modified to go through
 *          gxact->inredo entries that have not made it to disk.
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include <fcntl.h>
#include <sys/stat.h>
#include <time.h>
#include <unistd.h>

#include "access/commit_ts.h"
#include "access/htup_details.h"
#include "access/subtrans.h"
#include "access/transam.h"
#include "access/twophase.h"
#include "access/twophase_rmgr.h"
#include "access/xact.h"
#include "access/xlog.h"
#include "access/xloginsert.h"
#include "access/xlogutils.h"
#include "access/xlogreader.h"
#include "catalog/pg_type.h"
#include "catalog/storage.h"
#include "catalog/pg_control.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "pg_trace.h"
#include "pgstat.h"
#include "replication/origin.h"
#include "replication/syncrep.h"
#include "replication/walsender.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/predicate.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/sinvaladt.h"
#include "storage/smgr.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/timestamp.h"
#ifdef PGXC
#include "pgxc/xc_maintenance_mode.h"
#include "pgxc/nodemgr.h"
#endif
#ifdef __SUPPORT_DISTRIBUTED_TRANSACTION__
#include "postmaster/autovacuum.h"
#include "utils/snapmgr.h"
#endif

#ifdef __TBASE__
#include "access/gtm.h"
#include "utils/timeout.h"
#include "utils/relcryptmap.h"
#endif
#include "pgxc/execRemote.h"


/*
 * Directory where Two-phase commit files reside within PGDATA
 */
#define TWOPHASE_DIR "pg_twophase"

#define TWOPHASE_RECORD_DIR "pg_2pc"

#define GET_2PC_FILE_PATH(path, tid) \
    snprintf(path, MAXPGPATH, TWOPHASE_RECORD_DIR "/%s", tid)

int  transaction_threshold = 200000;

#define FILE_CONTENT_SIZE 2048

/* GUC variable, can't be changed after startup */
#ifdef PGXC
int            max_prepared_xacts = 10000;  /* We require 2PC */
#else
int            max_prepared_xacts = 0;
#endif
#ifdef __TBASE__
bool        enable_2pc_recovery_info = true;
#endif

#ifdef __TWO_PHASE_TRANS__
static HTAB *record_2pc_cache = NULL;

bool enable_2pc_file_cache = true;
bool enable_2pc_file_check = false;
bool enable_2pc_entry_key_check = true;
bool enable_2pc_entry_trace = false;

int record_2pc_cache_size = 4096;
int record_2pc_entry_size = 2048;

#define MAX_OUTPUT_FILE 1000

#define MAX_TID_SIZE        MAXPGPATH
#define MAX_2PC_INFO_SIZE   (record_2pc_entry_size - MAX_TID_SIZE)
#define DFLT_2PC_INFO_SIZE  1024  /* default size */

uint32 Record2pcCacheHashCode(const char *tid);

/*
 * The 2pc info cache is partitioned to reduce contention.
 * To determine which partition lock a given tid requires, compute the tid's
 * hash code with Record2pcCacheHashCode(), then apply Cache2pcPartitionLock().
 * NB: NUM_CACHE_2PC_PARTITIONS must be a power of 2!
 */
#define Cache2pcHashPartition(hashcode) \
	((hashcode) % NUM_CACHE_2PC_PARTITIONS)
#define Cache2pcPartitionLock(hashcode) \
	(&MainLWLockArray[CACHE_2PC_LWLOCK_OFFSET + \
		Cache2pcHashPartition(hashcode)].lock)
#define Cache2pcPartitionLockByIndex(i) \
	(&MainLWLockArray[CACHE_2PC_LWLOCK_OFFSET + (i)].lock)

/* hash table entry for 2pc record */
typedef struct Cache2pcInfo
{
	char key[MAX_TID_SIZE];        /* hash key: tid */
	char info[DFLT_2PC_INFO_SIZE];

} Cache2pcInfo;

static void check_entry_key(const char *tid, const char *key);

bool add_2pc_info(const char *tid, const char *info);

bool append_2pc_info(const char *tid, const char *info, bool *overflow);

bool remove_2pc_info(const char *tid);

bool get_2pc_info(const char *tid, char *info);

bool save_and_remove_2pc_info(const char *tid);

void check_2pc_file(const char *tid, const char *info, const char *func);

#endif

static GlobalTransaction
LookupGXact(const char *gid, Oid user);

static GlobalTransaction
LockGXact(const char *gid, Oid user, bool is_check);



/*
 * This struct describes one global transaction that is in prepared state
 * or attempting to become prepared.
 *
 * The lifecycle of a global transaction is:
 *
 * 1. After checking that the requested GID is not in use, set up an entry in
 * the TwoPhaseState->prepXacts array with the correct GID and valid = false,
 * and mark it as locked by my backend.
 *
 * 2. After successfully completing prepare, set valid = true and enter the
 * referenced PGPROC into the global ProcArray.
 *
 * 3. To begin COMMIT PREPARED or ROLLBACK PREPARED, check that the entry is
 * valid and not locked, then mark the entry as locked by storing my current
 * backend ID into locking_backend.  This prevents concurrent attempts to
 * commit or rollback the same prepared xact.
 *
 * 4. On completion of COMMIT PREPARED or ROLLBACK PREPARED, remove the entry
 * from the ProcArray and the TwoPhaseState->prepXacts array and return it to
 * the freelist.
 *
 * Note that if the preparing transaction fails between steps 1 and 2, the
 * entry must be removed so that the GID and the GlobalTransaction struct
 * can be reused.  See AtAbort_Twophase().
 *
 * typedef struct GlobalTransactionData *GlobalTransaction appears in
 * twophase.h
 *
 * Note that the max value of GIDSIZE must fit in the uint16 gidlen,
 * specified in TwoPhaseFileHeader.
 * refer to MAX_IMPLICIT_2PC_STR_LEN (5 + 21 + 64 + 1 + 5 + 5), bigger than it.
 */

typedef struct GlobalTransactionData
{
    GlobalTransaction next;        /* list link for free list */
    int            pgprocno;        /* ID of associated dummy PGPROC */
    BackendId    dummyBackendId; /* similar to backend id for backends */
    TimestampTz prepared_at;    /* time of preparation */
#ifdef __SUPPORT_DISTRIBUTED_TRANSACTION__
    TimestampTz prepared_timestamp;    /* prepared global timestamp */
#endif
    /*
     * Note that we need to keep track of two LSNs for each GXACT. We keep
     * track of the start LSN because this is the address we must use to read
     * state data back from WAL when committing a prepared GXACT. We keep
     * track of the end LSN because that is the LSN we need to wait for prior
     * to commit.
     */
    XLogRecPtr    prepare_start_lsn;    /* XLOG offset of prepare record start */
    XLogRecPtr    prepare_end_lsn;    /* XLOG offset of prepare record end */
    TransactionId xid;            /* The GXACT id */

    Oid            owner;            /* ID of user that executed the xact */
    BackendId    locking_backend;    /* backend currently working on the xact */
    bool        valid;            /* TRUE if PGPROC entry is in proc array */
    bool        ondisk;            /* TRUE if prepare state file is on disk */
    bool        inredo;            /* TRUE if entry was added via xlog_redo */
    char        gid[GIDSIZE];    /* The GID assigned to the prepared xact */
}            GlobalTransactionData;

/*
 * Two Phase Commit shared state.  Access to this struct is protected
 * by TwoPhaseStateLock.
 */
typedef struct TwoPhaseStateData
{
    /* Head of linked list of free GlobalTransactionData structs */
    GlobalTransaction freeGXacts;

    /* Number of valid prepXacts entries. */
    int            numPrepXacts;

    /* There are max_prepared_xacts items in this array */
    GlobalTransaction prepXacts[FLEXIBLE_ARRAY_MEMBER];
} TwoPhaseStateData;

static TwoPhaseStateData *TwoPhaseState;

/*
 * Global transaction entry currently locked by us, if any.  Note that any
 * access to the entry pointed to by this variable must be protected by
 * TwoPhaseStateLock, though obviously the pointer itself doesn't need to be
 * (since it's just local memory).
 */
static GlobalTransaction MyLockedGxact = NULL;

static bool twophaseExitRegistered = false;

static void RecordTransactionCommitPrepared(TransactionId xid,
                                int nchildren,
                                TransactionId *children,
                                int nrels,
                                RelFileNode *rels,
                                int ninvalmsgs,
                                SharedInvalidationMessage *invalmsgs,
                                bool initfileinval);
static void RecordTransactionAbortPrepared(TransactionId xid,
                               int nchildren,
                               TransactionId *children,
                               int nrels,
                               RelFileNode *rels);
static void ProcessRecords(char *bufptr, TransactionId xid,
               const TwoPhaseCallback callbacks[]);
static void RemoveGXact(GlobalTransaction gxact);

static void XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len);
static char *ProcessTwoPhaseBuffer(TransactionId xid,
                      XLogRecPtr prepare_start_lsn,
                      bool fromdisk, bool setParent, bool setNextXid);
static void MarkAsPreparingGuts(GlobalTransaction gxact, TransactionId xid,
                    const char *gid, TimestampTz prepared_at, Oid owner,
                    Oid databaseid);
static void RemoveTwoPhaseFile(TransactionId xid, bool giveWarning);
static void RecreateTwoPhaseFile(TransactionId xid, void *content, int len);

/*
 * Initialization of shared memory
 */
Size
TwoPhaseShmemSize(void)
{
    Size        size;

    /* Need the fixed struct, the array of pointers, and the GTD structs */
    size = offsetof(TwoPhaseStateData, prepXacts);
    size = add_size(size, mul_size(max_prepared_xacts,
                                   sizeof(GlobalTransaction)));
    size = MAXALIGN(size);
    size = add_size(size, mul_size(max_prepared_xacts,
                                   sizeof(GlobalTransactionData)));

    return size;
}

void
TwoPhaseShmemInit(void)
{
    bool        found;

    TwoPhaseState = ShmemInitStruct("Prepared Transaction Table",
                                    TwoPhaseShmemSize(),
                                    &found);
    if (!IsUnderPostmaster)
    {
        GlobalTransaction gxacts;
        int            i;

        Assert(!found);
        TwoPhaseState->freeGXacts = NULL;
        TwoPhaseState->numPrepXacts = 0;

        /*
         * Initialize the linked list of free GlobalTransactionData structs
         */
        gxacts = (GlobalTransaction)
            ((char *) TwoPhaseState +
             MAXALIGN(offsetof(TwoPhaseStateData, prepXacts) +
                      sizeof(GlobalTransaction) * max_prepared_xacts));
        for (i = 0; i < max_prepared_xacts; i++)
        {
            /* insert into linked list */
            gxacts[i].next = TwoPhaseState->freeGXacts;
            TwoPhaseState->freeGXacts = &gxacts[i];

            /* associate it with a PGPROC assigned by InitProcGlobal */
            gxacts[i].pgprocno = PreparedXactProcs[i].pgprocno;

            /*
             * Assign a unique ID for each dummy proc, so that the range of
             * dummy backend IDs immediately follows the range of normal
             * backend IDs. We don't dare to assign a real backend ID to dummy
             * procs, because prepared transactions don't take part in cache
             * invalidation like a real backend ID would imply, but having a
             * unique ID for them is nevertheless handy. This arrangement
             * allows you to allocate an array of size (MaxBackends +
             * max_prepared_xacts + 1), and have a slot for every backend and
             * prepared transaction. Currently multixact.c uses that
             * technique.
             */
            gxacts[i].dummyBackendId = MaxBackends + 1 + i;
        }
    }
    else
        Assert(found);
}

/*
 * Exit hook to unlock the global transaction entry we're working on.
 */
static void
AtProcExit_Twophase(int code, Datum arg)
{
    /* same logic as abort */
    AtAbort_Twophase();
}

/*
 * Abort hook to unlock the global transaction entry we're working on.
 */
void
AtAbort_Twophase(void)
{
    if (MyLockedGxact == NULL)
        return;

    /*
     * What to do with the locked global transaction entry?  If we were in the
     * process of preparing the transaction, but haven't written the WAL
     * record and state file yet, the transaction must not be considered as
     * prepared.  Likewise, if we are in the process of finishing an
     * already-prepared transaction, and fail after having already written the
     * 2nd phase commit or rollback record to the WAL, the transaction should
     * not be considered as prepared anymore.  In those cases, just remove the
     * entry from shared memory.
     *
     * Otherwise, the entry must be left in place so that the transaction can
     * be finished later, so just unlock it.
     *
     * If we abort during prepare, after having written the WAL record, we
     * might not have transferred all locks and other state to the prepared
     * transaction yet.  Likewise, if we abort during commit or rollback,
     * after having written the WAL record, we might not have released all the
     * resources held by the transaction yet.  In those cases, the in-memory
     * state can be wrong, but it's too late to back out.
     */
    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    if (!MyLockedGxact->valid)
        RemoveGXact(MyLockedGxact);
    else
        MyLockedGxact->locking_backend = InvalidBackendId;
    LWLockRelease(TwoPhaseStateLock);

    MyLockedGxact = NULL;
}

/*
 * This is called after we have finished transferring state to the prepared
 * PGXACT entry.
 */
void
PostPrepare_Twophase(void)
{
    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    MyLockedGxact->locking_backend = InvalidBackendId;
    LWLockRelease(TwoPhaseStateLock);

    MyLockedGxact = NULL;
}

#ifdef __SUPPORT_DISTRIBUTED_TRANSACTION__

static void
RecoverEndGlobalPrepare(GlobalTransaction gxact)
{
    volatile PGXACT    *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
    
    pg_atomic_write_u64(&pgxact->prepare_timestamp, gxact->prepared_timestamp);
    elog(LOG, "recover proc no %d prepare timestamp " INT64_FORMAT " xid %d.", gxact->pgprocno, 
                            gxact->prepared_timestamp, pgxact->xid);

}

void
EndGlobalPrepare(GlobalTransaction gxact, bool isImplicit)
{	
	volatile PGXACT	*pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
	
	pg_atomic_write_u64(&pgxact->prepare_timestamp, GetGlobalPrepareTimestamp());
	if(enable_distri_print)
	{
		elog(LOG, "proc no %d prepare timestamp " INT64_FORMAT " xid %d.", gxact->pgprocno, 
							GetGlobalPrepareTimestamp(), pgxact->xid);
	}

    if(isImplicit && !GlobalTimestampIsValid(pg_atomic_read_u64(&pgxact->prepare_timestamp)))
    {
        elog(ERROR, "prepare transaction %d does not have valid prepare timestamp. autovacuum %d", 
             gxact->xid, IsAutoVacuumWorkerProcess());
    }

#ifdef __TBASE_DEBUG__
	if(enable_distri_print)
	{
		InsertPreparedXid(pgxact->xid, GetGlobalPrepareTimestamp());
	}
#endif	

	// SetGlobalPrepareTimestamp(InvalidGlobalTimestamp);
	/* 
	 * Transfer the tmin to the prepared proc without locking. 
	 * As the prepare xact procs lie behind the normal procs in proc array,
	 * Get Snapshot would not miss the tmin even when it is being transferred.
	 * 
	 * According to PortalRunUtility, we do not set snapshot if transaction
	 * only contains utilities that do not need one. In that case,
	 * xmin and tmin are both invalid, for they are both set by snapshot.
	 * So if xmin is valid, tmin should also be.
	 */
	pg_atomic_write_u64(&pgxact->tmin, pg_atomic_read_u64(&MyPgXact->tmin));
	if(!GlobalTimestampIsValid(pg_atomic_read_u64(&MyPgXact->tmin)) &&
		TransactionIdIsValid(MyPgXact->xmin))
	{
		elog(LOG,
			 "prepare transaction %d does not have valid tmin. autovacuum %d", 
			 MyPgXact->xid, IsAutoVacuumWorkerProcess());
	}
}


void EndExplicitGlobalPrepare(char *gid)
{
    GlobalTransaction gxact;
    volatile PGXACT       *pgxact;
    
    gxact = LookupGXact(gid, GetUserId());
    if(NULL == gxact)
    {
        return;
    }
    
    pgxact =  &ProcGlobal->allPgXact[gxact->pgprocno];
    pg_atomic_write_u64(&pgxact->prepare_timestamp, GetGlobalPrepareTimestamp());
    if(enable_distri_print)
    {
        elog(LOG, "proc no %d explicit prepare timestamp " INT64_FORMAT " xid %d gid %s.", 
                        gxact->pgprocno, GetGlobalPrepareTimestamp(), GetTopTransactionIdIfAny(), gid);
    }
    if(!GlobalTimestampIsValid(pg_atomic_read_u64(&pgxact->prepare_timestamp)))
    {
        ereport(ERROR,
            (errcode(ERRCODE_INTERNAL_ERROR),
             errmsg("explicit prepared transaction %d does not have valid prepare timestamp. autovacuum %d", 
             gxact->xid, IsAutoVacuumWorkerProcess())));
    }
    SetGlobalPrepareTimestamp(InvalidGlobalTimestamp);
}
#endif



/*
 * MarkAsPreparing
 *        Reserve the GID for the given transaction.
 */
GlobalTransaction
MarkAsPreparing(TransactionId xid, const char *gid,
                TimestampTz prepared_at, Oid owner, Oid databaseid)
{
    GlobalTransaction gxact;
    int            i;

    if (strlen(gid) >= GIDSIZE)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("transaction identifier \"%s\" is too long",
                        gid)));

    /* fail immediately if feature is disabled */
    if (max_prepared_xacts == 0)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("prepared transactions are disabled"),
                 errhint("Set max_prepared_transactions to a nonzero value.")));

    /* on first call, register the exit hook */
    if (!twophaseExitRegistered)
    {
        before_shmem_exit(AtProcExit_Twophase, 0);
        twophaseExitRegistered = true;
    }

    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);

    /* Check for conflicting GID */
    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        gxact = TwoPhaseState->prepXacts[i];
        if (strcmp(gxact->gid, gid) == 0)
        {
            ereport(ERROR,
                    (errcode(ERRCODE_DUPLICATE_OBJECT),
                     errmsg("transaction identifier \"%s\" is already in use",
                            gid)));
        }
    }

    /* Get a free gxact from the freelist */
    if (TwoPhaseState->freeGXacts == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_OUT_OF_MEMORY),
                 errmsg("maximum number of prepared transactions reached"),
                 errhint("Increase max_prepared_transactions (currently %d).",
                         max_prepared_xacts)));
    gxact = TwoPhaseState->freeGXacts;
    TwoPhaseState->freeGXacts = gxact->next;

    MarkAsPreparingGuts(gxact, xid, gid, prepared_at, owner, databaseid);

    gxact->ondisk = false;

    /* And insert it into the active array */
    Assert(TwoPhaseState->numPrepXacts < max_prepared_xacts);
    TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts++] = gxact;

    LWLockRelease(TwoPhaseStateLock);

    return gxact;
}

/*
 * MarkAsPreparingGuts
 *
 * This uses a gxact struct and puts it into the active array.
 * NOTE: this is also used when reloading a gxact after a crash; so avoid
 * assuming that we can use very much backend context.
 *
 * Note: This function should be called with appropriate locks held.
 */
static void
MarkAsPreparingGuts(GlobalTransaction gxact, TransactionId xid, const char *gid,
                    TimestampTz prepared_at, Oid owner, Oid databaseid)
{
    PGPROC       *proc;
    PGXACT       *pgxact;
    int            i;

    Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));

    Assert(gxact != NULL);
    proc = &ProcGlobal->allProcs[gxact->pgprocno];
    pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];

    /* Initialize the PGPROC entry */
    MemSet(proc, 0, sizeof(PGPROC));
    proc->pgprocno = gxact->pgprocno;
    SHMQueueElemInit(&(proc->links));
    proc->waitStatus = STATUS_OK;
    /* We set up the gxact's VXID as InvalidBackendId/XID */
    proc->lxid = (LocalTransactionId) xid;
    pgxact->xid = xid;
    pgxact->xmin = InvalidTransactionId;
    pgxact->delayChkpt = false;
    pgxact->vacuumFlags = 0;
    proc->pid = 0;
#ifdef __SUPPORT_DISTRIBUTED_TRANSACTION__
    pg_atomic_init_u64(&pgxact->prepare_timestamp, InvalidGlobalTimestamp);
    pg_atomic_init_u64(&pgxact->tmin, InvalidGlobalTimestamp);
    proc->hasGlobalXid = false;
#endif
    proc->backendId = InvalidBackendId;
    proc->databaseId = databaseid;
    proc->roleId = owner;
    proc->isBackgroundWorker = false;
    proc->lwWaiting = false;
    proc->lwWaitMode = 0;
    proc->waitLock = NULL;
    proc->waitProcLock = NULL;
    for (i = 0; i < NUM_LOCK_PARTITIONS; i++)
        SHMQueueInit(&(proc->myProcLocks[i]));
    /* subxid data must be filled later by GXactLoadSubxactData */
    pgxact->overflowed = false;
    pgxact->nxids = 0;

    gxact->prepared_at = prepared_at;
    gxact->xid = xid;
    gxact->owner = owner;
    gxact->locking_backend = MyBackendId;
    gxact->valid = false;
    gxact->inredo = false;
    strcpy(gxact->gid, gid);

    /*
     * Remember that we have this GlobalTransaction entry locked for us. If we
     * abort after this, we must release it.
     */
    MyLockedGxact = gxact;
}

/*
 * GXactLoadSubxactData
 *
 * If the transaction being persisted had any subtransactions, this must
 * be called before MarkAsPrepared() to load information into the dummy
 * PGPROC.
 */
static void
GXactLoadSubxactData(GlobalTransaction gxact, int nsubxacts,
                     TransactionId *children)
{
    PGPROC       *proc = &ProcGlobal->allProcs[gxact->pgprocno];
    PGXACT       *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];

    /* We need no extra lock since the GXACT isn't valid yet */
    if (nsubxacts > PGPROC_MAX_CACHED_SUBXIDS)
    {
        pgxact->overflowed = true;
        nsubxacts = PGPROC_MAX_CACHED_SUBXIDS;
    }
    if (nsubxacts > 0)
    {
        memcpy(proc->subxids.xids, children,
               nsubxacts * sizeof(TransactionId));
        pgxact->nxids = nsubxacts;
    }
}

/*
 * MarkAsPrepared
 *        Mark the GXACT as fully valid, and enter it into the global ProcArray.
 *
 * lock_held indicates whether caller already holds TwoPhaseStateLock.
 */
static void
MarkAsPrepared(GlobalTransaction gxact, bool lock_held)
{
    /* Lock here may be overkill, but I'm not convinced of that ... */
    if (!lock_held)
        LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    Assert(!gxact->valid);
    gxact->valid = true;
    if (!lock_held)
        LWLockRelease(TwoPhaseStateLock);


    /*
     * Put it into the global ProcArray so TransactionIdIsInProgress considers
     * the XID as still running.
     */
    ProcArrayAdd(&ProcGlobal->allProcs[gxact->pgprocno]);
}

static GlobalTransaction
LookupGXact(const char *gid, Oid user)
{// #lizard forgives
    int            i;


    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);

    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
        PGPROC       *proc = &ProcGlobal->allProcs[gxact->pgprocno];

        /* Ignore not-yet-valid GIDs */
        if (!gxact->valid)
            continue;
        if (strcmp(gxact->gid, gid) != 0)
            continue;

        /* Found it, but has someone else got it locked? */
        if (gxact->locking_backend != InvalidBackendId)
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                     errmsg("prepared transaction with identifier \"%s\" is busy",
                            gid)));

        if (user != gxact->owner && !superuser_arg(user))
            ereport(ERROR,
                    (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                     errmsg("permission denied to finish prepared transaction"),
                     errhint("Must be superuser or the user that prepared the transaction.")));

        /*
         * Note: it probably would be possible to allow committing from
         * another database; but at the moment NOTIFY is known not to work and
         * there may be some other issues as well.  Hence disallow until
         * someone gets motivated to make it work.
         */
        if (MyDatabaseId != proc->databaseId)
            ereport(ERROR,
                    (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                     errmsg("prepared transaction belongs to another database"),
                     errhint("Connect to the database where the transaction was prepared to finish it.")));

        LWLockRelease(TwoPhaseStateLock);

        return gxact;
    }

    LWLockRelease(TwoPhaseStateLock);

#ifdef PGXC
    /*
     * In PGXC, if xc_maintenance_mode is on, COMMIT/ROLLBACK PREPARED may be issued to the
     * node where the given xid does not exist.
     */
    if (!xc_maintenance_mode)
    {
#endif
    ereport(ERROR,
            (errcode(ERRCODE_UNDEFINED_OBJECT),
             errmsg("prepared transaction with identifier \"%s\" does not exist",
                    gid)));
#ifdef PGXC
    }
#endif

    /* NOTREACHED */
    return NULL;
}


/*
 * LockGXact
 *        Locate the prepared transaction and mark it busy for COMMIT or PREPARE.
 */
static GlobalTransaction
LockGXact(const char *gid, Oid user, bool is_check)
{// #lizard forgives
    int            i;

    /* on first call, register the exit hook */
    if (!twophaseExitRegistered)
    {
        before_shmem_exit(AtProcExit_Twophase, 0);
        twophaseExitRegistered = true;
    }

    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);

    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
        PGPROC       *proc = &ProcGlobal->allProcs[gxact->pgprocno];

        /* Ignore not-yet-valid GIDs */
        if (!gxact->valid)
            continue;
        if (strcmp(gxact->gid, gid) != 0)
            continue;

        /* Found it, but has someone else got it locked? */
        if (gxact->locking_backend != InvalidBackendId)
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                     errmsg("prepared transaction with identifier \"%s\" is busy",
                            gid)));

        if (user != gxact->owner && !superuser_arg(user))
            ereport(ERROR,
                    (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                     errmsg("permission denied to finish prepared transaction"),
                     errhint("Must be superuser or the user that prepared the transaction.")));

        /*
         * Note: it probably would be possible to allow committing from
         * another database; but at the moment NOTIFY is known not to work and
         * there may be some other issues as well.  Hence disallow until
         * someone gets motivated to make it work.
         */
        if (MyDatabaseId != proc->databaseId)
            ereport(ERROR,
                    (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                     errmsg("prepared transaction belongs to another database"),
                     errhint("Connect to the database where the transaction was prepared to finish it.")));

		if (!is_check)
		{
        /* OK for me to lock it */
        gxact->locking_backend = MyBackendId;
        MyLockedGxact = gxact;
		}

        LWLockRelease(TwoPhaseStateLock);

        return gxact;
    }

    LWLockRelease(TwoPhaseStateLock);

#ifdef PGXC
    /*
     * In PGXC, if xc_maintenance_mode is on, COMMIT/ROLLBACK PREPARED may be issued to the
     * node where the given xid does not exist.
     */
    if (!xc_maintenance_mode)
    {
#endif
    ereport(ERROR,
            (errcode(ERRCODE_UNDEFINED_OBJECT),
             errmsg("prepared transaction with identifier \"%s\" does not exist",
                    gid)));
#ifdef PGXC
    }
#endif

    /* NOTREACHED */
    return NULL;
}

/*
 * RemoveGXact
 *        Remove the prepared transaction from the shared memory array.
 *
 * NB: caller should have already removed it from ProcArray
 */
static void
RemoveGXact(GlobalTransaction gxact)
{
    int            i;

    Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));

    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        if (gxact == TwoPhaseState->prepXacts[i])
        {
            /* remove from the active array */
            TwoPhaseState->numPrepXacts--;
            TwoPhaseState->prepXacts[i] = TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts];

            /* and put it back in the freelist */
            gxact->next = TwoPhaseState->freeGXacts;
            TwoPhaseState->freeGXacts = gxact;

            return;
        }
    }

    elog(ERROR, "failed to find %p in GlobalTransaction array", gxact);
}

/*
 * Returns an array of all prepared transactions for the user-level
 * function pg_prepared_xact.
 *
 * The returned array and all its elements are copies of internal data
 * structures, to minimize the time we need to hold the TwoPhaseStateLock.
 *
 * WARNING -- we return even those transactions that are not fully prepared
 * yet.  The caller should filter them out if he doesn't want them.
 *
 * The returned array is palloc'd.
 */
static int
GetPreparedTransactionList(GlobalTransaction *gxacts)
{
    GlobalTransaction array;
    int            num;
    int            i;

    LWLockAcquire(TwoPhaseStateLock, LW_SHARED);

    if (TwoPhaseState->numPrepXacts == 0)
    {
        LWLockRelease(TwoPhaseStateLock);

        *gxacts = NULL;
        return 0;
    }

    num = TwoPhaseState->numPrepXacts;
    array = (GlobalTransaction) palloc(sizeof(GlobalTransactionData) * num);
    *gxacts = array;
    for (i = 0; i < num; i++)
        memcpy(array + i, TwoPhaseState->prepXacts[i],
               sizeof(GlobalTransactionData));

    LWLockRelease(TwoPhaseStateLock);

    return num;
}


/* Working status for pg_prepared_xact */
typedef struct
{
    GlobalTransaction array;
    int            ngxacts;
    int            currIdx;
} Working_State;

/*
 * pg_prepared_xact
 *        Produce a view with one row per prepared transaction.
 *
 * This function is here so we don't have to export the
 * GlobalTransactionData struct definition.
 */
Datum
pg_prepared_xact(PG_FUNCTION_ARGS)
{
    FuncCallContext *funcctx;
    Working_State *status;

    if (SRF_IS_FIRSTCALL())
    {
        TupleDesc    tupdesc;
        MemoryContext oldcontext;

        /* create a function context for cross-call persistence */
        funcctx = SRF_FIRSTCALL_INIT();

        /*
         * Switch to memory context appropriate for multiple function calls
         */
        oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);

        /* build tupdesc for result tuples */
        /* this had better match pg_prepared_xacts view in system_views.sql */
        tupdesc = CreateTemplateTupleDesc(5, false);
        TupleDescInitEntry(tupdesc, (AttrNumber) 1, "transaction",
                           XIDOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber) 2, "gid",
                           TEXTOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber) 3, "prepared",
                           TIMESTAMPTZOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber) 4, "ownerid",
                           OIDOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber) 5, "dbid",
                           OIDOID, -1, 0);

        funcctx->tuple_desc = BlessTupleDesc(tupdesc);

        /*
         * Collect all the 2PC status information that we will format and send
         * out as a result set.
         */
        status = (Working_State *) palloc(sizeof(Working_State));
        funcctx->user_fctx = (void *) status;

        status->ngxacts = GetPreparedTransactionList(&status->array);
        status->currIdx = 0;

        MemoryContextSwitchTo(oldcontext);
    }

    funcctx = SRF_PERCALL_SETUP();
    status = (Working_State *) funcctx->user_fctx;

    while (status->array != NULL && status->currIdx < status->ngxacts)
    {
        GlobalTransaction gxact = &status->array[status->currIdx++];
        PGPROC       *proc = &ProcGlobal->allProcs[gxact->pgprocno];
        PGXACT       *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
        Datum        values[5];
        bool        nulls[5];
        HeapTuple    tuple;
        Datum        result;

        if (!gxact->valid)
            continue;

        /*
         * Form tuple with appropriate data.
         */
        MemSet(values, 0, sizeof(values));
        MemSet(nulls, 0, sizeof(nulls));

        values[0] = TransactionIdGetDatum(pgxact->xid);
        values[1] = CStringGetTextDatum(gxact->gid);
        values[2] = TimestampTzGetDatum(gxact->prepared_at);
        values[3] = ObjectIdGetDatum(gxact->owner);
        values[4] = ObjectIdGetDatum(proc->databaseId);

        tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
        result = HeapTupleGetDatum(tuple);
        SRF_RETURN_NEXT(funcctx, result);
    }

    SRF_RETURN_DONE(funcctx);
}

/*
 * TwoPhaseGetGXact
 *        Get the GlobalTransaction struct for a prepared transaction
 *        specified by XID
 */
static GlobalTransaction
TwoPhaseGetGXact(TransactionId xid)
{
    GlobalTransaction result = NULL;
    int            i;

    static TransactionId cached_xid = InvalidTransactionId;
    static GlobalTransaction cached_gxact = NULL;

    /*
     * During a recovery, COMMIT PREPARED, or ABORT PREPARED, we'll be called
     * repeatedly for the same XID.  We can save work with a simple cache.
     */
    if (xid == cached_xid)
        return cached_gxact;

    LWLockAcquire(TwoPhaseStateLock, LW_SHARED);

    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
        PGXACT       *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];

        if (pgxact->xid == xid)
        {
            result = gxact;
            break;
        }
    }

    LWLockRelease(TwoPhaseStateLock);

    if (result == NULL)            /* should not happen */
        elog(ERROR, "failed to find GlobalTransaction for xid %u", xid);

    cached_xid = xid;
    cached_gxact = result;

    return result;
}

/*
 * TwoPhaseGetDummyProc
 *        Get the dummy backend ID for prepared transaction specified by XID
 *
 * Dummy backend IDs are similar to real backend IDs of real backends.
 * They start at MaxBackends + 1, and are unique across all currently active
 * real backends and prepared transactions.
 */
BackendId
TwoPhaseGetDummyBackendId(TransactionId xid)
{
    GlobalTransaction gxact = TwoPhaseGetGXact(xid);

    return gxact->dummyBackendId;
}

/*
 * TwoPhaseGetDummyProc
 *        Get the PGPROC that represents a prepared transaction specified by XID
 */
PGPROC *
TwoPhaseGetDummyProc(TransactionId xid)
{
    GlobalTransaction gxact = TwoPhaseGetGXact(xid);

    return &ProcGlobal->allProcs[gxact->pgprocno];
}

/************************************************************************/
/* State file support                                                    */
/************************************************************************/

#define TwoPhaseFilePath(path, xid) \
    snprintf(path, MAXPGPATH, TWOPHASE_DIR "/%08X", xid)

/*
 * 2PC state file format:
 *
 *    1. TwoPhaseFileHeader
 *    2. TransactionId[] (subtransactions)
 *    3. RelFileNode[] (files to be deleted at commit)
 *    4. RelFileNode[] (files to be deleted at abort)
 *    5. SharedInvalidationMessage[] (inval messages to be sent at commit)
 *    6. TwoPhaseRecordOnDisk
 *    7. ...
 *    8. TwoPhaseRecordOnDisk (end sentinel, rmid == TWOPHASE_RM_END_ID)
 *    9. checksum (CRC-32C)
 *
 * Each segment except the final checksum is MAXALIGN'd.
 */

/*
 * Header for a 2PC state file
 */
#define TWOPHASE_MAGIC    0x57F94533    /* format identifier */

typedef struct TwoPhaseFileHeader
{
    uint32        magic;            /* format identifier */
    uint32        total_len;        /* actual file length */
    TransactionId xid;            /* original transaction XID */
    Oid            database;        /* OID of database it was in */
    TimestampTz prepared_at;    /* time of preparation */
    Oid            owner;            /* user running the transaction */
    int32        nsubxacts;        /* number of following subxact XIDs */
    int32        ncommitrels;    /* number of delete-on-commit rels */
    int32        nabortrels;        /* number of delete-on-abort rels */
    int32        ninvalmsgs;        /* number of cache invalidation messages */
#ifdef __SUPPORT_DISTRIBUTED_TRANSACTION__
    TimestampTz prepared_timestamp; /* global prepare timestamp */
#endif
#ifdef __TBASE__
    int32       ncreatesegs;    /* number of created sequences */
    int32       ndropsegs;        /* number of dropped sequences */
    int32       nrenamesegs;    /* number of renamed sequences */
#endif
    bool        initfileinval;    /* does relcache init file need invalidation? */
    uint16        gidlen;            /* length of the GID - GID follows the header */
} TwoPhaseFileHeader;

/*
 * Header for each record in a state file
 *
 * NOTE: len counts only the rmgr data, not the TwoPhaseRecordOnDisk header.
 * The rmgr data will be stored starting on a MAXALIGN boundary.
 */
typedef struct TwoPhaseRecordOnDisk
{
    uint32        len;            /* length of rmgr data */
    TwoPhaseRmgrId rmid;        /* resource manager for this record */
    uint16        info;            /* flag bits for use by rmgr */
} TwoPhaseRecordOnDisk;

/*
 * During prepare, the state file is assembled in memory before writing it
 * to WAL and the actual state file.  We use a chain of StateFileChunk blocks
 * for that.
 */
typedef struct StateFileChunk
{
    char       *data;
    uint32        len;
    struct StateFileChunk *next;
} StateFileChunk;

static struct xllist
{
    StateFileChunk *head;        /* first data block in the chain */
    StateFileChunk *tail;        /* last block in chain */
    uint32        num_chunks;
    uint32        bytes_free;        /* free bytes left in tail block */
    uint32        total_len;        /* total data bytes in chain */
}            records;


/*
 * Append a block of data to records data structure.
 *
 * NB: each block is padded to a MAXALIGN multiple.  This must be
 * accounted for when the file is later read!
 *
 * The data is copied, so the caller is free to modify it afterwards.
 */
static void
save_state_data(const void *data, uint32 len)
{
    uint32        padlen = MAXALIGN(len);

    if (padlen > records.bytes_free)
    {
        records.tail->next = palloc0(sizeof(StateFileChunk));
        records.tail = records.tail->next;
        records.tail->len = 0;
        records.tail->next = NULL;
        records.num_chunks++;

        records.bytes_free = Max(padlen, 512);
        records.tail->data = palloc(records.bytes_free);
    }

    memcpy(((char *) records.tail->data) + records.tail->len, data, len);
    records.tail->len += padlen;
    records.bytes_free -= padlen;
    records.total_len += padlen;
}

/*
 * Start preparing a state file.
 *
 * Initializes data structure and inserts the 2PC file header record.
 */
void
StartPrepare(GlobalTransaction gxact)
{// #lizard forgives
    PGPROC       *proc = &ProcGlobal->allProcs[gxact->pgprocno];
    PGXACT       *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
    TransactionId xid = pgxact->xid;
    TwoPhaseFileHeader hdr;
    TransactionId *children;
    RelFileNode *commitrels;
    RelFileNode *abortrels;
    SharedInvalidationMessage *invalmsgs;
#ifdef __TBASE__
    char *create_info = NULL;
    char *drop_info   = NULL;
    char *rename_info = NULL;
#endif
    /* Initialize linked list */
    records.head = palloc0(sizeof(StateFileChunk));
    records.head->len = 0;
    records.head->next = NULL;

    records.bytes_free = Max(sizeof(TwoPhaseFileHeader), 512);
    records.head->data = palloc(records.bytes_free);

    records.tail = records.head;
    records.num_chunks = 1;

    records.total_len = 0;

    /* Create header */
    hdr.magic = TWOPHASE_MAGIC;
    hdr.total_len = 0;            /* EndPrepare will fill this in */
    hdr.xid = xid;
    hdr.database = proc->databaseId;
    hdr.prepared_at = gxact->prepared_at;
    hdr.owner = gxact->owner;
    hdr.nsubxacts = xactGetCommittedChildren(&children);
    hdr.ncommitrels = smgrGetPendingDeletes(true, &commitrels);
    hdr.nabortrels = smgrGetPendingDeletes(false, &abortrels);
    hdr.ninvalmsgs = xactGetCommittedInvalidationMessages(&invalmsgs,
                                                          &hdr.initfileinval);
#ifdef __TBASE__
    hdr.ncreatesegs = GetGTMCreateSeq(&create_info);
    hdr.ndropsegs   = GetGTMDropSeq(&drop_info);
    hdr.nrenamesegs    = GetGTMRenameSeq(&rename_info);
#endif
#ifdef __SUPPORT_DISTRIBUTED_TRANSACTION__
    hdr.prepared_timestamp = GetGlobalPrepareTimestamp();
    if(enable_distri_print)
    {
        elog(LOG, "2pc record prepare timestamp "INT64_FORMAT, hdr.prepared_timestamp);
    }
#endif
    hdr.gidlen = strlen(gxact->gid) + 1;    /* Include '\0' */

    save_state_data(&hdr, sizeof(TwoPhaseFileHeader));
    save_state_data(gxact->gid, hdr.gidlen);

    /*
     * Add the additional info about subxacts, deletable files and cache
     * invalidation messages.
     */
    if (hdr.nsubxacts > 0)
    {
        save_state_data(children, hdr.nsubxacts * sizeof(TransactionId));
        /* While we have the child-xact data, stuff it in the gxact too */
        GXactLoadSubxactData(gxact, hdr.nsubxacts, children);
    }
    if (hdr.ncommitrels > 0)
    {
        save_state_data(commitrels, hdr.ncommitrels * sizeof(RelFileNode));
        pfree(commitrels);
    }
    if (hdr.nabortrels > 0)
    {
        save_state_data(abortrels, hdr.nabortrels * sizeof(RelFileNode));
        pfree(abortrels);
    }
    if (hdr.ninvalmsgs > 0)
    {
        save_state_data(invalmsgs,
                        hdr.ninvalmsgs * sizeof(SharedInvalidationMessage));
        pfree(invalmsgs);
    }
    
#ifdef __TBASE__
    if (hdr.ncreatesegs > 0)
    {
        save_state_data(create_info,
                        hdr.ncreatesegs * sizeof(CreateInfo));
        pfree(create_info);
    }

    if (hdr.ndropsegs > 0)
    {
        save_state_data(drop_info,
                        hdr.ndropsegs * sizeof(DropInfo));
        pfree(drop_info);
    }

    if (hdr.nrenamesegs > 0)
    {
        save_state_data(rename_info,
                        hdr.nrenamesegs * sizeof(RenameInfo));
        pfree(rename_info);
    }
#endif
}

/*
 * Finish preparing state data and writing it to WAL.
 */
void
EndPrepare(GlobalTransaction gxact)
{// #lizard forgives
    TwoPhaseFileHeader *hdr;
    StateFileChunk *record;

    /*
     * Here is where we really truly prepare.
     *
     * We have to record transaction prepares even if we didn't make any
     * updates, because the transaction manager might get confused if we lose
     * a global transaction.
     */
#ifdef __TWO_PHASE_TRANS__
    if (enable_distri_print)
    {
        elog(LOG, "record remote 2pc on node:%s,  gid: %s, startnode:%s, startxid: %u, "
                                     "partnodes:%s, localxid: %u", PGXCNodeName, gxact->gid, 
                                         g_twophase_state.start_node_name, 
                                         g_twophase_state.start_xid, 
                                         g_twophase_state.participants, 
                                         gxact->xid);
    }

    /*
     * record 2pc file for explicit trans at the beginning of EndPrepare,
     * since g_twophase_state is cleared after the explicit "PREPARE TRANSACTION"
     */
    if (!g_twophase_state.is_readonly)
    {
#ifdef __TWO_PHASE_TESTS__
        if (g_twophase_state.is_start_node && 
            (FILE_KERNEL_EXSISTED == twophase_exception_case || 
                (FILE_PGCLEAN_EXISTED_CONSISTENT == twophase_exception_case && 
                    g_twophase_state.in_pg_clean)))
        {
            record_2pc_involved_nodes_xid(gxact->gid, 
                                  g_twophase_state.start_node_name, 
                                  g_twophase_state.start_xid, 
                                  g_twophase_state.participants, 
                                  gxact->xid);
        }
        if (g_twophase_state.is_start_node && 
            FILE_PGCLEAN_EXISTED_NCONSISTENT == twophase_exception_case &&
                g_twophase_state.in_pg_clean)
        {
            record_2pc_involved_nodes_xid(gxact->gid, 
                                      g_twophase_state.start_node_name, 
                                      g_twophase_state.start_xid, 
                                      g_twophase_state.participants, 
                                      gxact->xid-1);
        }
#endif
        record_2pc_involved_nodes_xid(gxact->gid, 
                                  g_twophase_state.start_node_name, 
                                  g_twophase_state.start_xid, 
                                  g_twophase_state.participants, 
                                  gxact->xid);
    }
    else
    {
        record_2pc_readonly(gxact->gid);
    }
    /* is a start node? */
    if (!IS_PGXC_LOCAL_COORDINATOR)
    {
        g_twophase_state.is_start_node = false;
        strncpy(g_twophase_state.gid, gxact->gid, GIDSIZE);
        g_twophase_state.state = TWO_PHASE_PREPARING;
    }
#endif    

#ifdef __TWO_PHASE_TESTS__
    if (PART_PREPARE_RESPONSE_ERROR == twophase_exception_case && 
        !g_twophase_state.is_start_node)
    {
        if (0 == (PGXCNodeId % 2))
        {
            complish = true;
            elog(ERROR, "PREPARE_ERROR_RESPONSE_ERROR at node '%s'", PGXCNodeName);
        }
    }
        
    if (g_twophase_state.is_start_node && 
        PART_PREPARE_ENDPREPARE == twophase_exception_case)
    {
        complish = true;
        elog(ERROR, "PART_PREPARE_ENDPREPARE complete");
    }
#endif

    /* Add the end sentinel to the list of 2PC records */
    RegisterTwoPhaseRecord(TWOPHASE_RM_END_ID, 0,
                           NULL, 0);

    /* Go back and fill in total_len in the file header record */
    hdr = (TwoPhaseFileHeader *) records.head->data;
    Assert(hdr->magic == TWOPHASE_MAGIC);
    hdr->total_len = records.total_len + sizeof(pg_crc32c);

    /*
     * If the data size exceeds MaxAllocSize, we won't be able to read it in
     * ReadTwoPhaseFile. Check for that now, rather than fail in the case
     * where we write data to file and then re-read at commit time.
     */
    if (hdr->total_len > MaxAllocSize)
        ereport(ERROR,
                (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                 errmsg("two-phase state file maximum length exceeded")));

    /*
     * Now writing 2PC state data to WAL. We let the WAL's CRC protection
     * cover us, so no need to calculate a separate CRC.
     *
     * We have to set delayChkpt here, too; otherwise a checkpoint starting
     * immediately after the WAL record is inserted could complete without
     * fsync'ing our state file.  (This is essentially the same kind of race
     * condition as the COMMIT-to-clog-write case that RecordTransactionCommit
     * uses delayChkpt for; see notes there.)
     *
     * We save the PREPARE record's location in the gxact for later use by
     * CheckPointTwoPhase.
     */
    XLogEnsureRecordSpace(0, records.num_chunks);

    START_CRIT_SECTION();

    MyPgXact->delayChkpt = true;

    XLogBeginInsert();
    for (record = records.head; record != NULL; record = record->next)
        XLogRegisterData(record->data, record->len);
    gxact->prepare_end_lsn = XLogInsert(RM_XACT_ID, XLOG_XACT_PREPARE);
    XLogFlush(gxact->prepare_end_lsn);

    /* If we crash now, we have prepared: WAL replay will fix things */

    /* Store record's start location to read that later on Commit */
    gxact->prepare_start_lsn = ProcLastRecPtr;

    /*
     * Mark the prepared transaction as valid.  As soon as xact.c marks
     * MyPgXact as not running our XID (which it will do immediately after
     * this function returns), others can commit/rollback the xact.
     *
     * NB: a side effect of this is to make a dummy ProcArray entry for the
     * prepared XID.  This must happen before we clear the XID from MyPgXact,
     * else there is a window where the XID is not running according to
     * TransactionIdIsInProgress, and onlookers would be entitled to assume
     * the xact crashed.  Instead we have a window where the same XID appears
     * twice in ProcArray, which is OK.
     */
    MarkAsPrepared(gxact, false);

    /*
     * Now we can mark ourselves as out of the commit critical section: a
     * checkpoint starting after this will certainly see the gxact as a
     * candidate for fsyncing.
     */
    MyPgXact->delayChkpt = false;

    /*
     * Remember that we have this GlobalTransaction entry locked for us.  If
     * we crash after this point, it's too late to abort, but we must unlock
     * it so that the prepared transaction can be committed or rolled back.
     */
    MyLockedGxact = gxact;

    END_CRIT_SECTION();

    /*
     * Wait for synchronous replication, if required.
     *
     * Note that at this stage we have marked the prepare, but still show as
     * running in the procarray (twice!) and continue to hold locks.
     */
    SyncRepWaitForLSN(gxact->prepare_end_lsn, false);

    records.tail = records.head = NULL;
    records.num_chunks = 0;
}

/*
 * Register a 2PC record to be written to state file.
 */
void
RegisterTwoPhaseRecord(TwoPhaseRmgrId rmid, uint16 info,
                       const void *data, uint32 len)
{
    TwoPhaseRecordOnDisk record;

    record.rmid = rmid;
    record.info = info;
    record.len = len;
    save_state_data(&record, sizeof(TwoPhaseRecordOnDisk));
    if (len > 0)
        save_state_data(data, len);
}


/*
 * Read and validate the state file for xid.
 *
 * If it looks OK (has a valid magic number and CRC), return the palloc'd
 * contents of the file.  Otherwise return NULL.
 */
static char *
ReadTwoPhaseFile(TransactionId xid, bool give_warnings)
{// #lizard forgives
    char        path[MAXPGPATH];
    char       *buf;
    TwoPhaseFileHeader *hdr;
    int            fd;
    struct stat stat;
    uint32        crc_offset;
    pg_crc32c    calc_crc,
                file_crc;

    TwoPhaseFilePath(path, xid);

    fd = OpenTransientFile(path, O_RDONLY | PG_BINARY, 0);
    if (fd < 0)
    {
        if (give_warnings)
            ereport(WARNING,
                    (errcode_for_file_access(),
                     errmsg("could not open two-phase state file \"%s\": %m",
                            path)));
        return NULL;
    }

    /*
     * Check file length.  We can determine a lower bound pretty easily. We
     * set an upper bound to avoid palloc() failure on a corrupt file, though
     * we can't guarantee that we won't get an out of memory error anyway,
     * even on a valid file.
     */
    if (fstat(fd, &stat))
    {
        CloseTransientFile(fd);
        if (give_warnings)
            ereport(WARNING,
                    (errcode_for_file_access(),
                     errmsg("could not stat two-phase state file \"%s\": %m",
                            path)));
        return NULL;
    }

    if (stat.st_size < (MAXALIGN(sizeof(TwoPhaseFileHeader)) +
                        MAXALIGN(sizeof(TwoPhaseRecordOnDisk)) +
                        sizeof(pg_crc32c)) ||
        stat.st_size > MaxAllocSize)
    {
        CloseTransientFile(fd);
        return NULL;
    }

    crc_offset = stat.st_size - sizeof(pg_crc32c);
    if (crc_offset != MAXALIGN(crc_offset))
    {
        CloseTransientFile(fd);
        return NULL;
    }

    /*
     * OK, slurp in the file.
     */
    buf = (char *) palloc(stat.st_size);

    pgstat_report_wait_start(WAIT_EVENT_TWOPHASE_FILE_READ);
    if (read(fd, buf, stat.st_size) != stat.st_size)
    {
        pgstat_report_wait_end();
        CloseTransientFile(fd);
        if (give_warnings)
            ereport(WARNING,
                    (errcode_for_file_access(),
                     errmsg("could not read two-phase state file \"%s\": %m",
                            path)));
        pfree(buf);
        return NULL;
    }

    pgstat_report_wait_end();
    CloseTransientFile(fd);

    hdr = (TwoPhaseFileHeader *) buf;
    if (hdr->magic != TWOPHASE_MAGIC || hdr->total_len != stat.st_size)
    {
        pfree(buf);
        return NULL;
    }

    INIT_CRC32C(calc_crc);
    COMP_CRC32C(calc_crc, buf, crc_offset);
    FIN_CRC32C(calc_crc);

    file_crc = *((pg_crc32c *) (buf + crc_offset));

    if (!EQ_CRC32C(calc_crc, file_crc))
    {
        pfree(buf);
        return NULL;
    }

    return buf;
}


/*
 * Reads 2PC data from xlog. During checkpoint this data will be moved to
 * twophase files and ReadTwoPhaseFile should be used instead.
 *
 * Note clearly that this function can access WAL during normal operation,
 * similarly to the way WALSender or Logical Decoding would do.
 *
 */
static void
XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len)
{
    XLogRecord *record;
    XLogReaderState *xlogreader;
    char       *errormsg;

    xlogreader = XLogReaderAllocate(&read_local_xlog_page, NULL);
    if (!xlogreader)
        ereport(ERROR,
                (errcode(ERRCODE_OUT_OF_MEMORY),
                 errmsg("out of memory"),
                 errdetail("Failed while allocating a WAL reading processor.")));

    record = XLogReadRecord(xlogreader, lsn, &errormsg);
    if (record == NULL)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not read two-phase state from WAL at %X/%X",
                        (uint32) (lsn >> 32),
                        (uint32) lsn)));

    if (XLogRecGetRmid(xlogreader) != RM_XACT_ID ||
        (XLogRecGetInfo(xlogreader) & XLOG_XACT_OPMASK) != XLOG_XACT_PREPARE)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("expected two-phase state data is not present in WAL at %X/%X",
                        (uint32) (lsn >> 32),
                        (uint32) lsn)));

    if (len != NULL)
        *len = XLogRecGetDataLen(xlogreader);

    *buf = palloc(sizeof(char) * XLogRecGetDataLen(xlogreader));
    memcpy(*buf, XLogRecGetData(xlogreader), sizeof(char) * XLogRecGetDataLen(xlogreader));

    XLogReaderFree(xlogreader);
}


/*
 * Confirms an xid is prepared, during recovery
 */
bool
StandbyTransactionIdIsPrepared(TransactionId xid)
{
    char       *buf;
    TwoPhaseFileHeader *hdr;
    bool        result;

    Assert(TransactionIdIsValid(xid));

    if (max_prepared_xacts <= 0)
        return false;            /* nothing to do */

    /* Read and validate file */
    buf = ReadTwoPhaseFile(xid, false);
    if (buf == NULL)
        return false;

    /* Check header also */
    hdr = (TwoPhaseFileHeader *) buf;
    result = TransactionIdEquals(hdr->xid, xid);
    pfree(buf);

    return result;
}

/*
 * CheckPreparedTransactionLock: Check whether the prepared transaction
 * can be rollbacked
 */
void
CheckPreparedTransactionLock(const char *gid)
{
	GlobalTransaction gxact = LockGXact(gid, GetUserId(), true);
	if (enable_distri_print)
	{
		if (gxact == NULL)
		{
			elog(LOG, "prepared gid %s gxact is NULL.", gid);
		}
		else
		{
			elog(LOG, "prepared gid %s gxact xid %d.", gid, gxact->xid);
		}
	}
}

/*
 * FinishPreparedTransaction: execute COMMIT PREPARED or ROLLBACK PREPARED
 */
void
FinishPreparedTransaction(const char *gid, bool isCommit)
{// #lizard forgives
    GlobalTransaction gxact;
    PGPROC       *proc;
    PGXACT       *pgxact;
    TransactionId xid;
    char       *buf;
    char       *bufptr;
    TwoPhaseFileHeader *hdr;
    TransactionId latestXid;
    TransactionId *children;
    RelFileNode *commitrels;
    RelFileNode *abortrels;
    RelFileNode *delrels;
    int            ndelrels;
    SharedInvalidationMessage *invalmsgs;
#ifdef __TBASE__
    RenameInfo  *rename_info = NULL;
    CreateInfo    *create_info = NULL;
    DropInfo    *drop_info   = NULL;
#endif
    int            i;


#ifdef __TWO_PHASE_TRANS__
    /* not allow user commit twophase trans in xc_maintenance_mode */
    if (xc_maintenance_mode)
    {
        elog(ERROR, "can not commit transaction '%s' in xc_maintainence_mode", gid);
    }
    /* after abnormal exit g_twophase_state is lost */
    if ('\0' == g_twophase_state.gid[0])
    {
        strncpy(g_twophase_state.gid, gid, GIDSIZE);
    }
    
    if (isCommit)
    {
        /* 
         *if current node is startnode, the twophase state should satisfy COMMIT_END in most situations, 
         *that is, remote nodes are already committed, or
         *explict trans exit after FinishGTMGID, or nodestring and gxid is lost due to GTM restart, or
         *the transaction abnormal exit last time and it lost g_twophase_state
         */
#ifdef __TBASE__
        HOLD_INTERRUPTS();
        disable_timeout_safely();
#endif
        if (g_twophase_state.is_start_node)
        {
            g_twophase_state.state = TWO_PHASE_COMMIT_END;
        }
        else
        {
            /* 
             *if current node is participate, the twophase state should be COMMITTING, 
             */
            g_twophase_state.state = TWO_PHASE_COMMITTING;
            /* explicit twophase state is cleaned after 'prepare transaction' */
            if (!IsXidImplicit(gid))
            {
                strncpy(g_twophase_state.gid, gid, GIDSIZE);
            }
        }
    }
    else
    {
        if (g_twophase_state.is_start_node)
        {
            g_twophase_state.state = TWO_PHASE_ABORT_END;
        }
        else
        {
            g_twophase_state.state = TWO_PHASE_ABORTTING;
            if (!IsXidImplicit(gid))
            {
                strncpy(g_twophase_state.gid, gid, GIDSIZE);
            }
        }
    }

#endif    

#ifdef __TWO_PHASE_TESTS__
    if (!g_twophase_state.is_start_node &&
        (PREPARE_ERROR_RESPONSE_ERROR == twophase_exception_case ||
            PART_COMMIT_RESPONSE_ERROR == twophase_exception_case))
    {
        if (0 == (PGXCNodeId % 2))
        {
            elog(ERROR, "%s at node '%s'", 
                (PREPARE_ERROR_RESPONSE_ERROR == twophase_exception_case) ? "PREPARE_ERROR_RESPONSE_ERROR" : "PART_COMMIT_RESPONSE_ERROR", 
                PGXCNodeName);
        }
    }
    if (g_twophase_state.is_start_node && 
        (PART_COMMIT_FINISH_PREPARED == twophase_exception_case))
    {
        elog(ERROR, "PART_COMMIT_FINISH_PREPARED at node '%s'", PGXCNodeName);
    }
#endif
    
    elog(DEBUG8, "finish prepared gid %s is commit %d.", gid, isCommit);
    /*
     * Validate the GID, and lock the GXACT to ensure that two backends do not
     * try to commit the same GID at once.
     */
	gxact = LockGXact(gid, GetUserId(), false);
#ifdef PGXC
    /*
     * LockGXact returns NULL if this node does not contain given two-phase
     * TXN.  This can happen when COMMIT/ROLLBACK PREPARED is issued at
     * the originating Coordinator for cleanup.
     * In this case, no local handling is needed.   Only report to GTM
     * is needed and this has already been handled in
     * FinishRemotePreparedTransaction().
     *
     * Second predicate may not be necessary.   It is just in case.
     */
    if (gxact == NULL && xc_maintenance_mode)
        return;
#endif
    if(enable_distri_print)
    {
        elog(LOG, "finish prepared gid %s is commit %d gxact xid %d.", gid, isCommit, gxact->xid);
    }
    proc = &ProcGlobal->allProcs[gxact->pgprocno];
    pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
    xid = pgxact->xid;

#ifdef __TWO_PHASE_TRANS__
    /* 
     * except the start node, other participants record 2pc file 
     * at the beginning of FinishPrepareTransaction , 
     * or record 2pc in start node during AbortTransaction, 
     * since other participants may not abortted
     */
     /*
      * if run in pg_clean, partial prepare transaction may rollback without write 2pc, since no participants
      * since partial prepare do not have enough infomation for record
      */
    if (enable_distri_print)
    {
        if (g_twophase_state.in_pg_clean)
        {
            elog(LOG, "before record 2pc of '%s', readonly:'%s', after_prepare:'%s'", 
                gid, (g_twophase_state.is_readonly) ? "true" : "false", (g_twophase_state.is_after_prepare) ? "true" : "false");
        }
    }

    
    if (!g_twophase_state.is_start_node && 
        !g_twophase_state.is_readonly && 
        isCommit && 
        (!g_twophase_state.in_pg_clean ||
            (g_twophase_state.in_pg_clean && 
            g_twophase_state.is_after_prepare)))
    {
        if (g_twophase_state.in_pg_clean)
        {
            record_2pc_involved_nodes_xid(gid, g_twophase_state.start_node_name, g_twophase_state.start_xid, g_twophase_state.participants, xid);
        }
        record_2pc_commit_timestamp(gid, GetGlobalCommitTimestamp());
    }

#endif

    /*
     * Read and validate 2PC state data. State data will typically be stored
     * in WAL files if the LSN is after the last checkpoint record, or moved
     * to disk if for some reason they have lived for a long time.
     */
    if (gxact->ondisk)
	{
        buf = ReadTwoPhaseFile(xid, true);
		if (NULL == buf)
		{
			ereport(PANIC,
					(errcode_for_file_access(),
					errmsg("read two-phase file failed, gid: %s", gid)));
		}
	}
    else
	{
        XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, NULL);
		if (NULL == buf)
		{
			ereport(PANIC,
					(errcode_for_file_access(),
					errmsg("read two-phase data from xlog failed, gid: %s", gid)));
		}
	}


    /*
     * Disassemble the header area
     */
    hdr = (TwoPhaseFileHeader *) buf;
    Assert(TransactionIdEquals(hdr->xid, xid));
    bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
    bufptr += MAXALIGN(hdr->gidlen);
    children = (TransactionId *) bufptr;
    bufptr += MAXALIGN(hdr->nsubxacts * sizeof(TransactionId));
    commitrels = (RelFileNode *) bufptr;
    bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileNode));
    abortrels = (RelFileNode *) bufptr;
    bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileNode));
    invalmsgs = (SharedInvalidationMessage *) bufptr;
    bufptr += MAXALIGN(hdr->ninvalmsgs * sizeof(SharedInvalidationMessage));


#ifdef __TBASE__
    create_info = (CreateInfo *) bufptr;
    bufptr += MAXALIGN(hdr->ncreatesegs * sizeof(CreateInfo));

    drop_info = (DropInfo *) bufptr;
    bufptr += MAXALIGN(hdr->ndropsegs * sizeof(DropInfo));

    rename_info = (RenameInfo *) bufptr;
    bufptr += MAXALIGN(hdr->nrenamesegs * sizeof(RenameInfo));

    RestoreSeqCreate(create_info, hdr->ncreatesegs);
    RestoreSeqDrop(drop_info, hdr->ndropsegs);
    RestoreSeqRename(rename_info, hdr->nrenamesegs);
#endif

    /* compute latestXid among all children */
    latestXid = TransactionIdLatest(xid, hdr->nsubxacts, children);

    /*
     * The order of operations here is critical: make the XLOG entry for
     * commit or abort, then mark the transaction committed or aborted in
     * pg_xact, then remove its PGPROC from the global ProcArray (which means
     * TransactionIdIsInProgress will stop saying the prepared xact is in
     * progress), then run the post-commit or post-abort callbacks. The
     * callbacks will release the locks the transaction held.
     */
    if (isCommit)
        RecordTransactionCommitPrepared(xid,
                                        hdr->nsubxacts, children,
                                        hdr->ncommitrels, commitrels,
                                        hdr->ninvalmsgs, invalmsgs,
                                        hdr->initfileinval);
    else
        RecordTransactionAbortPrepared(xid,
                                       hdr->nsubxacts, children,
                                       hdr->nabortrels, abortrels);

    ProcArrayRemove(proc, latestXid);

    /*
     * In case we fail while running the callbacks, mark the gxact invalid so
     * no one else will try to commit/rollback, and so it will be recycled if
     * we fail after this point.  It is still locked by our backend so it
     * won't go away yet.
     *
     * (We assume it's safe to do this without taking TwoPhaseStateLock.)
     */
    gxact->valid = false;

    /*
     * We have to remove any files that were supposed to be dropped. For
     * consistency with the regular xact.c code paths, must do this before
     * releasing locks, so do it before running the callbacks.
     *
     * NB: this code knows that we couldn't be dropping any temp rels ...
     */
    if (isCommit)
    {
        delrels = commitrels;
        ndelrels = hdr->ncommitrels;
    }
    else
    {
        delrels = abortrels;
        ndelrels = hdr->nabortrels;
    }
    for (i = 0; i < ndelrels; i++)
    {
        SMgrRelation srel = smgropen(delrels[i], InvalidBackendId);

        smgrdounlink(srel, false);
#ifdef _MLS_
		/*
		 * clean up the rnode infomation in rel crypt hash table
		 */
		remove_rel_crypt_hash_elem(&(srel->smgr_relcrypt), true);
#endif
        smgrclose(srel);
    }

    /*
     * Handle cache invalidation messages.
     *
     * Relcache init file invalidation requires processing both before and
     * after we send the SI messages. See AtEOXact_Inval()
     */
    if (hdr->initfileinval)
        RelationCacheInitFilePreInvalidate();
    SendSharedInvalidMessages(invalmsgs, hdr->ninvalmsgs);
    if (hdr->initfileinval)
        RelationCacheInitFilePostInvalidate();

    /* And now do the callbacks */
    if (isCommit)
        ProcessRecords(bufptr, xid, twophase_postcommit_callbacks);
    else
        ProcessRecords(bufptr, xid, twophase_postabort_callbacks);

    
#ifdef __TBASE__
    FinishSeqOp(isCommit);
#endif

    PredicateLockTwoPhaseFinish(xid, isCommit);

    /* Count the prepared xact as committed or aborted */
    AtEOXact_PgStat(isCommit);

    /*
     * And now we can clean up any files we may have left.
     */
    if (gxact->ondisk)
        RemoveTwoPhaseFile(xid, true);

    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    RemoveGXact(gxact);
    LWLockRelease(TwoPhaseStateLock);
    MyLockedGxact = NULL;    

    pfree(buf);
#ifdef __TBASE__
    
    if(enable_distri_debug)
    {
        if(enable_distri_print)
        {
            elog(LOG, "finish prepared transaction top xid %u", xid);
        }
        LogCommitTranGTM(xid, 
                         gid, 
                         PGXCNodeName, 
                         1, 
                         false,
                         isCommit, 
                         GetLocalPrepareTimestamp(), 
                         GetLocalCommitTimestamp());
    }
    
    AtEOXact_Global();
#ifdef __TBASE_DEBUG__
    if(enable_distri_print)
    {
        DeletePreparedXid(xid);
    }
#endif    

#endif

#ifdef __TWO_PHASE_TRANS__
    /*
     * explain the code from two angles:
     * a. startnode: whole twophase trans complete 'commit prepared' or 'rollback prepared'
     * b. participates: complete ''commit prepared' or 'rollback prepared' only on local node 
     */
    g_twophase_state.state = (isCommit == true) ? TWO_PHASE_COMMITTED : TWO_PHASE_ABORTTED;

    /* delete the 2pc file while twophase transaction is committed on the current node */
    if(isCommit)
    {
        remove_2pc_records(gid, false);
    }
    else
    {
        /* rename 2pc file when rollback on the current node */
        rename_2pc_records(gid, 0);
    }

    ClearLocalTwoPhaseState();
    
    if (isCommit) 
    {
        RESUME_INTERRUPTS();
        /* clean timeout status */

        if (IsQueryCancelPending())
        {
            QueryCancelPending =  false;
            InterruptPending  = false;
        }            
    }
#endif    

}

/*
 * Scan 2PC state data in memory and call the indicated callbacks for each 2PC record.
 */
static void
ProcessRecords(char *bufptr, TransactionId xid,
               const TwoPhaseCallback callbacks[])
{
    for (;;)
    {
        TwoPhaseRecordOnDisk *record = (TwoPhaseRecordOnDisk *) bufptr;

        Assert(record->rmid <= TWOPHASE_RM_MAX_ID);
        if (record->rmid == TWOPHASE_RM_END_ID)
            break;

        bufptr += MAXALIGN(sizeof(TwoPhaseRecordOnDisk));

        if (callbacks[record->rmid] != NULL)
            callbacks[record->rmid] (xid, record->info,
                                     (void *) bufptr, record->len);

        bufptr += MAXALIGN(record->len);
    }
}

/*
 * Remove the 2PC file for the specified XID.
 *
 * If giveWarning is false, do not complain about file-not-present;
 * this is an expected case during WAL replay.
 */
static void
RemoveTwoPhaseFile(TransactionId xid, bool giveWarning)
{
    char        path[MAXPGPATH];

    TwoPhaseFilePath(path, xid);
    if (unlink(path))
        if (errno != ENOENT || giveWarning)
            ereport(WARNING,
                    (errcode_for_file_access(),
                     errmsg("could not remove two-phase state file \"%s\": %m",
                            path)));
}

/*
 * Recreates a state file. This is used in WAL replay and during
 * checkpoint creation.
 *
 * Note: content and len don't include CRC.
 */
static void
RecreateTwoPhaseFile(TransactionId xid, void *content, int len)
{
    char        path[MAXPGPATH];
    pg_crc32c    statefile_crc;
    int            fd;

    /* Recompute CRC */
    INIT_CRC32C(statefile_crc);
    COMP_CRC32C(statefile_crc, content, len);
    FIN_CRC32C(statefile_crc);

    TwoPhaseFilePath(path, xid);

    fd = OpenTransientFile(path,
                           O_CREAT | O_TRUNC | O_WRONLY | PG_BINARY,
                           S_IRUSR | S_IWUSR);
    if (fd < 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not recreate two-phase state file \"%s\": %m",
                        path)));

    /* Write content and CRC */
    pgstat_report_wait_start(WAIT_EVENT_TWOPHASE_FILE_WRITE);
    if (write(fd, content, len) != len)
    {
        pgstat_report_wait_end();
        CloseTransientFile(fd);
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not write two-phase state file: %m")));
    }
    if (write(fd, &statefile_crc, sizeof(pg_crc32c)) != sizeof(pg_crc32c))
    {
        pgstat_report_wait_end();
        CloseTransientFile(fd);
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not write two-phase state file: %m")));
    }
    pgstat_report_wait_end();

    /*
     * We must fsync the file because the end-of-replay checkpoint will not do
     * so, there being no GXACT in shared memory yet to tell it to.
     */
    pgstat_report_wait_start(WAIT_EVENT_TWOPHASE_FILE_SYNC);
    if (pg_fsync(fd) != 0)
    {
        CloseTransientFile(fd);
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not fsync two-phase state file: %m")));
    }
    pgstat_report_wait_end();

    if (CloseTransientFile(fd) != 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not close two-phase state file: %m")));
}

/*
 * CheckPointTwoPhase -- handle 2PC component of checkpointing.
 *
 * We must fsync the state file of any GXACT that is valid or has been
 * generated during redo and has a PREPARE LSN <= the checkpoint's redo
 * horizon.  (If the gxact isn't valid yet, has not been generated in
 * redo, or has a later LSN, this checkpoint is not responsible for
 * fsyncing it.)
 *
 * This is deliberately run as late as possible in the checkpoint sequence,
 * because GXACTs ordinarily have short lifespans, and so it is quite
 * possible that GXACTs that were valid at checkpoint start will no longer
 * exist if we wait a little bit. With typical checkpoint settings this
 * will be about 3 minutes for an online checkpoint, so as a result we
 * we expect that there will be no GXACTs that need to be copied to disk.
 *
 * If a GXACT remains valid across multiple checkpoints, it will already
 * be on disk so we don't bother to repeat that write.
 */
void
CheckPointTwoPhase(XLogRecPtr redo_horizon)
{// #lizard forgives
    int            i;
    int            serialized_xacts = 0;

	elog(LOG, "[%s] checkpoint: "UINT64_FORMAT, __FUNCTION__, redo_horizon);

    if (max_prepared_xacts <= 0)
        return;                    /* nothing to do */

    TRACE_POSTGRESQL_TWOPHASE_CHECKPOINT_START();

    /*
     * We are expecting there to be zero GXACTs that need to be copied to
     * disk, so we perform all I/O while holding TwoPhaseStateLock for
     * simplicity. This prevents any new xacts from preparing while this
     * occurs, which shouldn't be a problem since the presence of long-lived
     * prepared xacts indicates the transaction manager isn't active.
     *
     * It's also possible to move I/O out of the lock, but on every error we
     * should check whether somebody committed our transaction in different
     * backend. Let's leave this optimization for future, if somebody will
     * spot that this place cause bottleneck.
     *
     * Note that it isn't possible for there to be a GXACT with a
     * prepare_end_lsn set prior to the last checkpoint yet is marked invalid,
     * because of the efforts with delayChkpt.
     */
    LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        /*
         * Note that we are using gxact not pgxact so this works in recovery
         * also
         */
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];

        if ((gxact->valid || gxact->inredo) &&
            gxact->prepare_end_lsn <= redo_horizon)
        {
            char       *buf;
            int            len;

			if (!gxact->ondisk)
			{
			/* save to pg_twophase */
            XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, &len);
            RecreateTwoPhaseFile(gxact->xid, buf, len);
			pfree(buf);

				gxact->ondisk = true;
				gxact->prepare_start_lsn = InvalidXLogRecPtr;
				gxact->prepare_end_lsn = InvalidXLogRecPtr;
				serialized_xacts++;
			}

#ifdef __TWO_PHASE_TRANS__
			/* save to pg_2pc */
			if (NULL != record_2pc_cache)
			{
				Assert(strlen(gxact->gid) < MAX_TID_SIZE);

				if (!save_and_remove_2pc_info(gxact->gid))
					{
					elog(LOG, "[%s] %s save to file failed",
						__FUNCTION__, gxact->gid);
				}
				else
				{
					elog(LOG, "[%s] %s is saved to file", __FUNCTION__, gxact->gid);
				}
			}
#endif
        }
    }
    LWLockRelease(TwoPhaseStateLock);

#ifdef __TWO_PHASE_TRANS__
	/* start node maybe no in prepared xacts */
	if (IS_PGXC_COORDINATOR && NULL != record_2pc_cache)
	{
		HASH_SEQ_STATUS seq;
		Cache2pcInfo *entry = NULL;
		char tid[MAX_TID_SIZE];
		char start_node[MAX_TID_SIZE];
		char *pos = NULL;
		int size = 0;

		/*
		 * set start_node likes ":cn001:"
		 * use to check whether the tid is started from this node
		 */
		memset(start_node, 0, MAX_TID_SIZE);
		size = strlen(PGXCNodeName);
		if (size + 2 >= MAX_TID_SIZE)
		{
			elog(PANIC, "[%s] node name length(%d) overflow", __FUNCTION__, size);
		}
		start_node[0] = ':';
		memcpy(start_node + 1, PGXCNodeName, size);
		start_node[size + 1] = ':';

		hash_seq_init(&seq, record_2pc_cache);
		while ((entry = hash_seq_search(&seq)) != NULL)
		{
			Assert(NULL != entry);

			size = strlen(entry->key);
			Assert(size < MAX_TID_SIZE);
			if (0 == size)
			{
				elog(LOG, "[%s] entry key is empty", __FUNCTION__);
					continue;
				}

			memset(tid, 0, MAX_TID_SIZE);
			memcpy(tid, entry->key, size + 1);
			if (0 == strlen(tid))
					{
				elog(LOG, "[%s] tid is empty", __FUNCTION__);
						continue;
					}
			Assert(strlen(tid) < MAX_TID_SIZE);

			if (enable_2pc_file_check)
				{
				elog(LOG, "[%s] %s is found in hash table seq", __FUNCTION__, tid);
			}

			if (IsXidImplicit(tid))
			{
				pos = strstr(tid, start_node);
				if (NULL == pos)
			{
					elog(LOG, "[%s] %s is not on start node", __FUNCTION__, tid);
					continue;
			}

				elog(LOG, "[%s] %s is on start node", __FUNCTION__, tid);
			}

			if (!save_and_remove_2pc_info(tid))
			{
				elog(LOG, "[%s] %s save to file failed", __FUNCTION__, tid);
			}
			else
			{
				elog(LOG, "[%s] %s is saved to file", __FUNCTION__, tid);
			}
		}
	}
#endif

    /*
     * Flush unconditionally the parent directory to make any information
     * durable on disk.  Two-phase files could have been removed and those
     * removals need to be made persistent as well as any files newly created
     * previously since the last checkpoint.
     */
    fsync_fname(TWOPHASE_DIR, true);
#ifdef __TWO_PHASE_TRANS__    
    fsync_fname(TWOPHASE_RECORD_DIR, true);
#endif

    TRACE_POSTGRESQL_TWOPHASE_CHECKPOINT_DONE();

    if (log_checkpoints && serialized_xacts > 0)
        ereport(LOG,
                (errmsg_plural("%u two-phase state file was written "
                               "for a long-running prepared transaction",
                               "%u two-phase state files were written "
                               "for long-running prepared transactions",
                               serialized_xacts,
                               serialized_xacts)));
}

/*
 * restoreTwoPhaseData
 *
 * Scan pg_twophase and fill TwoPhaseState depending on the on-disk data.
 * This is called once at the beginning of recovery, saving any extra
 * lookups in the future.  Two-phase files that are newer than the
 * minimum XID horizon are discarded on the way.
 */
void
restoreTwoPhaseData(void)
{
    DIR           *cldir;
    struct dirent *clde;

    cldir = AllocateDir(TWOPHASE_DIR);
    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    while ((clde = ReadDir(cldir, TWOPHASE_DIR)) != NULL)
    {
        if (strlen(clde->d_name) == 8 &&
            strspn(clde->d_name, "0123456789ABCDEF") == 8)
        {
            TransactionId xid;
            char       *buf;

            xid = (TransactionId) strtoul(clde->d_name, NULL, 16);

            buf = ProcessTwoPhaseBuffer(xid, InvalidXLogRecPtr,
                                        true, false, false);
            if (buf == NULL)
                continue;

            PrepareRedoAdd(buf, InvalidXLogRecPtr, InvalidXLogRecPtr);
        }
    }
    LWLockRelease(TwoPhaseStateLock);
    FreeDir(cldir);
}

/*
 * PrescanPreparedTransactions
 *
 * Scan the shared memory entries of TwoPhaseState and determine the range
 * of valid XIDs present.  This is run during database startup, after we
 * have completed reading WAL.  ShmemVariableCache->nextXid has been set to
 * one more than the highest XID for which evidence exists in WAL.
 *
 * We throw away any prepared xacts with main XID beyond nextXid --- if any
 * are present, it suggests that the DBA has done a PITR recovery to an
 * earlier point in time without cleaning out pg_twophase.  We dare not
 * try to recover such prepared xacts since they likely depend on database
 * state that doesn't exist now.
 *
 * However, we will advance nextXid beyond any subxact XIDs belonging to
 * valid prepared xacts.  We need to do this since subxact commit doesn't
 * write a WAL entry, and so there might be no evidence in WAL of those
 * subxact XIDs.
 *
 * Our other responsibility is to determine and return the oldest valid XID
 * among the prepared xacts (if none, return ShmemVariableCache->nextXid).
 * This is needed to synchronize pg_subtrans startup properly.
 *
 * If xids_p and nxids_p are not NULL, pointer to a palloc'd array of all
 * top-level xids is stored in *xids_p. The number of entries in the array
 * is returned in *nxids_p.
 */
TransactionId
PrescanPreparedTransactions(TransactionId **xids_p, int *nxids_p)
{
    TransactionId origNextXid = ShmemVariableCache->nextXid;
    TransactionId result = origNextXid;
    TransactionId *xids = NULL;
    int            nxids = 0;
    int            allocsize = 0;
    int            i;

    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        TransactionId xid;
        char       *buf;
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];

        Assert(gxact->inredo);

        xid = gxact->xid;

        buf = ProcessTwoPhaseBuffer(xid,
                                    gxact->prepare_start_lsn,
                                    gxact->ondisk, false, true);

        if (buf == NULL)
            continue;

        /*
         * OK, we think this file is valid.  Incorporate xid into the
         * running-minimum result.
         */
        if (TransactionIdPrecedes(xid, result))
            result = xid;

        if (xids_p)
        {
            if (nxids == allocsize)
            {
                if (nxids == 0)
                {
                    allocsize = 10;
                    xids = palloc(allocsize * sizeof(TransactionId));
                }
                else
                {
                    allocsize = allocsize * 2;
                    xids = repalloc(xids, allocsize * sizeof(TransactionId));
                }
            }
            xids[nxids++] = xid;
        }

        pfree(buf);
    }
    LWLockRelease(TwoPhaseStateLock);

    if (xids_p)
    {
        *xids_p = xids;
        *nxids_p = nxids;
    }

    return result;
}

/*
 * StandbyRecoverPreparedTransactions
 *
 * Scan the shared memory entries of TwoPhaseState and setup all the required
 * information to allow standby queries to treat prepared transactions as still
 * active.
 *
 * This is never called at the end of recovery - we use
 * RecoverPreparedTransactions() at that point.
 *
 * The lack of calls to SubTransSetParent() calls here is by design;
 * those calls are made by RecoverPreparedTransactions() at the end of recovery
 * for those xacts that need this.
 */
void
StandbyRecoverPreparedTransactions(void)
{
    int            i;

    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        TransactionId xid;
        char       *buf;
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];

        Assert(gxact->inredo);

        xid = gxact->xid;

        buf = ProcessTwoPhaseBuffer(xid,
                                    gxact->prepare_start_lsn,
                                    gxact->ondisk, false, false);
        if (buf != NULL)
            pfree(buf);
    }
    LWLockRelease(TwoPhaseStateLock);
}

/*
 * RecoverPreparedTransactions
 *
 * Scan the shared memory entries of TwoPhaseState and reload the state for
 * each prepared transaction (reacquire locks, etc).
 *
 * This is run at the end of recovery, but before we allow backends to write
 * WAL.
 *
 * At the end of recovery the way we take snapshots will change. We now need
 * to mark all running transactions with their full SubTransSetParent() info
 * to allow normal snapshots to work correctly if snapshots overflow.
 * We do this here because by definition prepared transactions are the only
 * type of write transaction still running, so this is necessary and
 * complete.
 */
void
RecoverPreparedTransactions(void)
{
    int            i;

    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        TransactionId xid;
        char       *buf;
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
        char       *bufptr;
        TwoPhaseFileHeader *hdr;
        TransactionId *subxids;
        const char *gid;

        xid = gxact->xid;

        /*
         * Reconstruct subtrans state for the transaction --- needed because
         * pg_subtrans is not preserved over a restart.  Note that we are
         * linking all the subtransactions directly to the top-level XID;
         * there may originally have been a more complex hierarchy, but
         * there's no need to restore that exactly. It's possible that
         * SubTransSetParent has been set before, if the prepared transaction
         * generated xid assignment records.
         */
        buf = ProcessTwoPhaseBuffer(xid,
                                    gxact->prepare_start_lsn,
                                    gxact->ondisk, true, false);
        if (buf == NULL)
            continue;

        ereport(LOG,
                (errmsg("recovering prepared transaction %u from shared memory", xid)));

        hdr = (TwoPhaseFileHeader *) buf;
        Assert(TransactionIdEquals(hdr->xid, xid));
        bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
        gid = (const char *) bufptr;
        bufptr += MAXALIGN(hdr->gidlen);
        subxids = (TransactionId *) bufptr;
        bufptr += MAXALIGN(hdr->nsubxacts * sizeof(TransactionId));
        bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileNode));
        bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileNode));
        bufptr += MAXALIGN(hdr->ninvalmsgs * sizeof(SharedInvalidationMessage));

#ifdef __TBASE__
        bufptr += MAXALIGN(hdr->ncreatesegs * sizeof(CreateInfo));
        bufptr += MAXALIGN(hdr->ndropsegs * sizeof(DropInfo));
        bufptr += MAXALIGN(hdr->nrenamesegs * sizeof(RenameInfo));
#endif
        /*
         * Recreate its GXACT and dummy PGPROC. But, check whether it was
         * added in redo and already has a shmem entry for it.
         */
        MarkAsPreparingGuts(gxact, xid, gid,
                            hdr->prepared_at,
                            hdr->owner, hdr->database);

        /* recovered, so reset the flag for entries generated by redo */
        gxact->inredo = false;

        GXactLoadSubxactData(gxact, hdr->nsubxacts, subxids);
        if(enable_distri_print)
        {
            elog(LOG, "recovery prepared transaction %u", gxact->xid);
        }


        MarkAsPrepared(gxact, true);
#ifdef __SUPPORT_DISTRIBUTED_TRANSACTION__
        RecoverEndGlobalPrepare(gxact);
#endif
        LWLockRelease(TwoPhaseStateLock);

        /*
         * Recover other state (notably locks) using resource managers.
         */
        ProcessRecords(bufptr, xid, twophase_recover_callbacks);

        /*
         * Release locks held by the standby process after we process each
         * prepared transaction. As a result, we don't need too many
         * additional locks at any one time.
         */
        if (InHotStandby)
            StandbyReleaseLockTree(xid, hdr->nsubxacts, subxids);

        /*
         * We're done with recovering this transaction. Clear MyLockedGxact,
         * like we do in PrepareTransaction() during normal operation.
         */
        PostPrepare_Twophase();

        pfree(buf);

        LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    }

    LWLockRelease(TwoPhaseStateLock);
}

/*
 * ProcessTwoPhaseBuffer
 *
 * Given a transaction id, read it either from disk or read it directly
 * via shmem xlog record pointer using the provided "prepare_start_lsn".
 *
 * If setParent is true, set up subtransaction parent linkages.
 *
 * If setNextXid is true, set ShmemVariableCache->nextXid to the newest
 * value scanned.
 */
static char *
ProcessTwoPhaseBuffer(TransactionId xid,
                      XLogRecPtr prepare_start_lsn,
                      bool fromdisk,
                      bool setParent, bool setNextXid)
{// #lizard forgives
    TransactionId origNextXid = ShmemVariableCache->nextXid;
    TransactionId *subxids;
    char       *buf;
    TwoPhaseFileHeader *hdr;
    int            i;

    Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));

    if (!fromdisk)
        Assert(prepare_start_lsn != InvalidXLogRecPtr);

    /* Already processed? */
    if (TransactionIdDidCommit(xid) || TransactionIdDidAbort(xid))
    {
        if (fromdisk)
        {
            ereport(WARNING,
                    (errmsg("removing stale two-phase state file for \"%u\"",
                            xid)));
            RemoveTwoPhaseFile(xid, true);
        }
        else
        {
            ereport(WARNING,
                    (errmsg("removing stale two-phase state from shared memory for \"%u\"",
                            xid)));
            PrepareRedoRemove(xid, true);
        }
        return NULL;
    }

    /* Reject XID if too new */
    if (TransactionIdFollowsOrEquals(xid, origNextXid))
    {
        if (fromdisk)
        {
            ereport(WARNING,
                    (errmsg("removing future two-phase state file for \"%u\"",
                            xid)));
            RemoveTwoPhaseFile(xid, true);
        }
        else
        {
            ereport(WARNING,
                    (errmsg("removing future two-phase state from memory for \"%u\"",
                            xid)));
            PrepareRedoRemove(xid, true);
        }
        return NULL;
    }

    if (fromdisk)
    {
        /* Read and validate file */
        buf = ReadTwoPhaseFile(xid, true);
        if (buf == NULL)
        {
            ereport(WARNING,
                    (errmsg("removing corrupt two-phase state file for \"%u\"",
                            xid)));
            RemoveTwoPhaseFile(xid, true);
            return NULL;
        }
    }
    else
    {
        /* Read xlog data */
        XlogReadTwoPhaseData(prepare_start_lsn, &buf, NULL);
    }

    /* Deconstruct header */
    hdr = (TwoPhaseFileHeader *) buf;
    if (!TransactionIdEquals(hdr->xid, xid))
    {
        if (fromdisk)
        {
            ereport(WARNING,
                    (errmsg("removing corrupt two-phase state file for \"%u\"",
                            xid)));
            RemoveTwoPhaseFile(xid, true);
        }
        else
        {
            ereport(WARNING,
                    (errmsg("removing corrupt two-phase state from memory for \"%u\"",
                            xid)));
            PrepareRedoRemove(xid, true);
        }
        pfree(buf);
        return NULL;
    }

    /*
     * Examine subtransaction XIDs ... they should all follow main XID, and
     * they may force us to advance nextXid.
     */
    subxids = (TransactionId *) (buf +
                                 MAXALIGN(sizeof(TwoPhaseFileHeader)) +
                                 MAXALIGN(hdr->gidlen));
    for (i = 0; i < hdr->nsubxacts; i++)
    {
        TransactionId subxid = subxids[i];

        Assert(TransactionIdFollows(subxid, xid));

        /* update nextXid if needed */
        if (setNextXid &&
            TransactionIdFollowsOrEquals(subxid,
                                         ShmemVariableCache->nextXid))
        {
            /*
             * We don't expect anyone else to modify nextXid, hence we don't
             * need to hold a lock while examining it.  We still acquire the
             * lock to modify it, though, so we recheck.
             */
            LWLockAcquire(XidGenLock, LW_EXCLUSIVE);
            if (TransactionIdFollowsOrEquals(subxid,
                                             ShmemVariableCache->nextXid))
            {
                ShmemVariableCache->nextXid = subxid;
                TransactionIdAdvance(ShmemVariableCache->nextXid);
            }
            LWLockRelease(XidGenLock);
        }

        if (setParent)
            SubTransSetParent(subxid, xid);
    }

    return buf;
}


/*
 *    RecordTransactionCommitPrepared
 *
 * This is basically the same as RecordTransactionCommit (q.v. if you change
 * this function): in particular, we must set the delayChkpt flag to avoid a
 * race condition.
 *
 * We know the transaction made at least one XLOG entry (its PREPARE),
 * so it is never possible to optimize out the commit record.
 */
static void
RecordTransactionCommitPrepared(TransactionId xid,
                                int nchildren,
                                TransactionId *children,
                                int nrels,
                                RelFileNode *rels,
                                int ninvalmsgs,
                                SharedInvalidationMessage *invalmsgs,
                                bool initfileinval)
{// #lizard forgives
    XLogRecPtr    recptr;
    TimestampTz committs = GetCurrentTimestamp();
    bool        replorigin;
#ifdef __SUPPORT_DISTRIBUTED_TRANSACTION__
    GlobalTimestamp global_committs = GetGlobalCommitTimestamp();
#endif

#ifdef __SUPPORT_DISTRIBUTED_TRANSACTION__
    if(false == GlobalTimestampIsValid(global_committs))
    {
        ereport(ERROR,
            (errcode(ERRCODE_INTERNAL_ERROR),
             errmsg("cannot commit a prepared transaction xid %d with invalid committs",
                    xid)));

    }
    
    if(enable_distri_print)
    {
        elog(LOG, "record commit prepared transaction xid %d ts "INT64_FORMAT, xid, global_committs);
    }
#endif
    /*
     * Are we using the replication origins feature?  Or, in other words, are
     * we replaying remote actions?
     */
    replorigin = (replorigin_session_origin != InvalidRepOriginId &&
                  replorigin_session_origin != DoNotReplicateId);

    START_CRIT_SECTION();

    /* See notes in RecordTransactionCommit */
    MyPgXact->delayChkpt = true;

    /*
     * Emit the XLOG commit record. Note that we mark 2PC commits as
     * potentially having AccessExclusiveLocks since we don't know whether or
     * not they do.
     */

#ifdef __SUPPORT_DISTRIBUTED_TRANSACTION__
    MyProc->commitTs = global_committs;

    recptr = XactLogCommitRecord(global_committs,
                                 committs,
                                 nchildren, children, nrels, rels,
                                 ninvalmsgs, invalmsgs,
                                 initfileinval, false,
                                 MyXactFlags | XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK,
                                 xid);
    elog(DEBUG8, "record transaction commit %d.", xid);
#else
    recptr = XactLogCommitRecord(InvalidGlobalTimestamp,
                                 committs,
                                 nchildren, children, nrels, rels,
                                 ninvalmsgs, invalmsgs,
                                 initfileinval, false,
                                 MyXactFlags | XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK,
                                 xid);

#endif

    if (replorigin)
        /* Move LSNs forward for this replication origin */
        replorigin_session_advance(replorigin_session_origin_lsn,
                                   XactLastRecEnd);

    /*
     * Record commit timestamp.  The value comes from plain commit timestamp
     * if replorigin is not enabled, or replorigin already set a value for us
     * in replorigin_session_origin_timestamp otherwise.
     *
     * We don't need to WAL-log anything here, as the commit record written
     * above already contains the data.
     */
    if (!replorigin || replorigin_session_origin_timestamp == 0)
        replorigin_session_origin_timestamp = committs;

#ifndef __SUPPORT_DISTRIBUTED_TRANSACTION__
    TransactionTreeSetCommitTsData(xid, nchildren, children,
                                   InvalidGlobalTimestamp,
                                   replorigin_session_origin_timestamp,
                                   replorigin_session_origin, false, InvalidXLogRecPtr);
#endif
    /*
     * We don't currently try to sleep before flush here ... nor is there any
     * support for async commit of a prepared xact (the very idea is probably
     * a contradiction)
     */

    /* Flush XLOG to disk */
    XLogFlush(recptr);


    /* Mark the transaction committed in pg_xact */
    TransactionIdCommitTree(xid, nchildren, children);


    /* Checkpoint can proceed now */
    MyPgXact->delayChkpt = false;

    END_CRIT_SECTION();

#ifdef __SUPPORT_DISTRIBUTED_TRANSACTION__
    if(TransactionIdIsNormal(xid))
    {
        if(GlobalTimestampIsValid(global_committs) && !CommitTimestampIsLocal(global_committs))
        {
            TransactionTreeSetCommitTsData(xid, nchildren, children,
                                           global_committs,
                                           replorigin_session_origin_timestamp,
                                           replorigin_session_origin, false, InvalidXLogRecPtr);
            
            if(enable_distri_print)
            {
                elog(LOG, "commit prepared xid %d time " INT64_FORMAT, xid, global_committs);
            }
        }
        else 
        {
            ereport(ERROR,
                    (errcode(ERRCODE_INTERNAL_ERROR),
                     errmsg("Commit prepared: Global Timestamp is not set. Current XID = %d, Autovac = %d, global committs " INT64_FORMAT, 
                                xid, IsAutoVacuumWorkerProcess(), global_committs)));
        }
    }
    else 
    {
        ereport(ERROR,
                    (errcode(ERRCODE_INTERNAL_ERROR),
                     errmsg("Commit prepared: transaction should not be abnormal. Current XID = %d, Autovac = %d", 
                                xid, IsAutoVacuumWorkerProcess())));
    }
    
    if(enable_distri_debug)
    {
        SetLocalCommitTimestamp(GetCurrentTimestamp());
    }
    MyProc->committed = true;
#endif

    /*
     * Wait for synchronous replication, if required.
     *
     * Note that at this stage we have marked clog, but still show as running
     * in the procarray and continue to hold locks.
     */
    SyncRepWaitForLSN(recptr, true);
}

/*
 *    RecordTransactionAbortPrepared
 *
 * This is basically the same as RecordTransactionAbort.
 *
 * We know the transaction made at least one XLOG entry (its PREPARE),
 * so it is never possible to optimize out the abort record.
 */
static void
RecordTransactionAbortPrepared(TransactionId xid,
                               int nchildren,
                               TransactionId *children,
                               int nrels,
                               RelFileNode *rels)
{
    XLogRecPtr    recptr;
#ifdef __SUPPORT_DISTRIBUTED_TRANSACTION__
	GlobalTimestamp global_time = InvalidGlobalTimestamp;
#endif

    /*
     * Catch the scenario where we aborted partway through
     * RecordTransactionCommitPrepared ...
     */
    if (TransactionIdDidCommit(xid))
        elog(PANIC, "cannot abort transaction %u, it was already committed",
             xid);

#ifdef __SUPPORT_DISTRIBUTED_TRANSACTION__
    global_time = GetGlobalTimestampGTM();
    
	if(!GlobalTimestampIsValid(global_time))
	{
		/* During the transaction rollback phase, 
		 * it is not mandatory to obtain a global timestamp from the GTM.
		 * it is sufficient to simply print a log when falied to get GTS from GTM
		 */
		elog(LOG, "failed to get global timestamp for abort command");
    }
    MyProc->commitTs = global_time;
#endif


    START_CRIT_SECTION();

    /*
     * Emit the XLOG commit record. Note that we mark 2PC aborts as
     * potentially having AccessExclusiveLocks since we don't know whether or
     * not they do.
     */
#ifdef __SUPPORT_DISTRIBUTED_TRANSACTION__
    recptr = XactLogAbortRecord(global_time,
                                GetCurrentTimestamp(),
                                nchildren, children,
                                nrels, rels,
                                MyXactFlags | XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK,
                                xid);
#else
    recptr = XactLogAbortRecord(InvalidGlobalTimestamp,
                                GetCurrentTimestamp(),
                                nchildren, children,
                                nrels, rels,
                                MyXactFlags | XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK,
                                xid);
#endif

    /* Always flush, since we're about to remove the 2PC state file */
    XLogFlush(recptr);

    /*
     * Mark the transaction aborted in clog.  This is not absolutely necessary
     * but we may as well do it while we are here.
     */
    TransactionIdAbortTree(xid, nchildren, children);

    END_CRIT_SECTION();

    /*
     * Wait for synchronous replication, if required.
     *
     * Note that at this stage we have marked clog, but still show as running
     * in the procarray and continue to hold locks.
     */
    SyncRepWaitForLSN(recptr, false);
}

/*
 * PrepareRedoAdd
 *
 * Store pointers to the start/end of the WAL record along with the xid in
 * a gxact entry in shared memory TwoPhaseState structure.  If caller
 * specifies InvalidXLogRecPtr as WAL location to fetch the two-phase
 * data, the entry is marked as located on disk.
 */
void
PrepareRedoAdd(char *buf, XLogRecPtr start_lsn, XLogRecPtr end_lsn)
{
    TwoPhaseFileHeader *hdr = (TwoPhaseFileHeader *) buf;
    char       *bufptr;
    const char *gid;
    GlobalTransaction gxact;

    Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
    Assert(RecoveryInProgress());

    bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
    gid = (const char *) bufptr;

    /*
     * Reserve the GID for the given transaction in the redo code path.
     *
     * This creates a gxact struct and puts it into the active array.
     *
     * In redo, this struct is mainly used to track PREPARE/COMMIT entries in
     * shared memory. Hence, we only fill up the bare minimum contents here.
     * The gxact also gets marked with gxact->inredo set to true to indicate
     * that it got added in the redo phase
     */

    /* Get a free gxact from the freelist */
    if (TwoPhaseState->freeGXacts == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_OUT_OF_MEMORY),
                 errmsg("maximum number of prepared transactions reached"),
                 errhint("Increase max_prepared_transactions (currently %d).",
                         max_prepared_xacts)));
    gxact = TwoPhaseState->freeGXacts;
    TwoPhaseState->freeGXacts = gxact->next;

    gxact->prepared_at = hdr->prepared_at;
    gxact->prepare_start_lsn = start_lsn;
    gxact->prepare_end_lsn = end_lsn;
    gxact->xid = hdr->xid;
    gxact->owner = hdr->owner;
    gxact->locking_backend = InvalidBackendId;
    gxact->valid = false;
    gxact->ondisk = XLogRecPtrIsInvalid(start_lsn);
    gxact->inredo = true;        /* yes, added in redo */
    strcpy(gxact->gid, gid);
#ifdef __SUPPORT_DISTRIBUTED_TRANSACTION__
    gxact->prepared_timestamp = hdr->prepared_timestamp;
    if(enable_distri_print)
    {
        elog(LOG, "2pc redo xid %d prepare timestamp "INT64_FORMAT, gxact->xid, gxact->prepared_timestamp);
    }
#endif

    /* And insert it into the active array */
    Assert(TwoPhaseState->numPrepXacts < max_prepared_xacts);
    TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts++] = gxact;

    elog(DEBUG2, "added 2PC data in shared memory for transaction %u", gxact->xid);
}

/*
 * PrepareRedoRemove
 *
 * Remove the corresponding gxact entry from TwoPhaseState. Also remove
 * the 2PC file if a prepared transaction was saved via an earlier checkpoint.
 *
 * Caller must hold TwoPhaseStateLock in exclusive mode, because TwoPhaseState
 * is updated.
 */
void
PrepareRedoRemove(TransactionId xid, bool giveWarning)
{
    GlobalTransaction gxact = NULL;
    int            i;
    bool        found = false;

    Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
    Assert(RecoveryInProgress());

    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        gxact = TwoPhaseState->prepXacts[i];

        if (gxact->xid == xid)
        {
            Assert(gxact->inredo);
            found = true;
            break;
        }
    }

    /*
     * Just leave if there is nothing, this is expected during WAL replay.
     */
    if (!found)
        return;

    /*
     * And now we can clean up any files we may have left.
     */
    elog(DEBUG2, "removing 2PC data for transaction %u", xid);
    if (gxact->ondisk)
        RemoveTwoPhaseFile(xid, giveWarning);
    RemoveGXact(gxact);

    return;
}

#ifdef __TWO_PHASE_TRANS__
/*
 * check_entry_key: check the entry key in the hash table whether is same with tid.
 */
static inline void check_entry_key(const char *tid, const char *key)
{
	if (enable_2pc_entry_key_check)
	{
	if (0 != strcmp(tid, key))
	{
			elog(PANIC, "%s(hashvalue: 0x%x) mismatch with %s(hashvalue: 0x%x)",
				tid, Record2pcCacheHashCode(tid), key, Record2pcCacheHashCode(key));
		}
	}
}

/*
 * add_2pc_info: add 2pc info to hash table
 * return true: add success
 * return false: add failed
 */
bool add_2pc_info(const char *tid, const char *info)
	{
	bool found = false;
		Cache2pcInfo *entry = NULL;
	uint32 hashvalue = Record2pcCacheHashCode(tid);
	LWLock *lock = Cache2pcPartitionLock(hashvalue);

	Assert(NULL != record_2pc_cache);
	Assert(NULL != tid);
	Assert(NULL != info);
	Assert(strlen(info) < MAX_2PC_INFO_SIZE);

	LWLockAcquire(lock, LW_EXCLUSIVE);

	entry = (Cache2pcInfo *)hash_search_with_hash_value(record_2pc_cache,
		tid, hashvalue, HASH_ENTER_NULL, &found);
	if (NULL == entry)
		{
		LWLockRelease(lock);
		return false;
	}

			Assert(NULL != entry);
	check_entry_key(tid, entry->key);

	memcpy(entry->info, info, strlen(info) + 1);

	LWLockRelease(lock);

	if (found)
	{
		elog(WARNING, "[%s] found %s", __FUNCTION__, tid);
		return true;
		}

	if (enable_2pc_entry_trace)
	{
		elog(LOG, "[%s] %s is added to hash table, entry: %p, info: %s",
			__FUNCTION__, tid, entry, info);
	}

	return true;
}

/*
 * append_2pc_info: append 2pc info to hash table
 * return true: append success
 * return false: append failed
 */
bool append_2pc_info(const char *tid, const char *info, bool *overflow)
	{
		bool found = false;
	int cur_size = 0;
	int app_size = 0;
	int new_size = 0;
	Cache2pcInfo *entry = NULL;
	uint32 hashvalue = Record2pcCacheHashCode(tid);
	LWLock *lock = Cache2pcPartitionLock(hashvalue);

	Assert(NULL != record_2pc_cache);
		Assert (NULL != tid);
		Assert (NULL != info);
	Assert(NULL != overflow);
	Assert(strlen(info) < MAX_2PC_INFO_SIZE);

	*overflow = false;

	LWLockAcquire(lock, LW_EXCLUSIVE);

	entry = (Cache2pcInfo *)hash_search_with_hash_value(record_2pc_cache,
		tid, hashvalue, HASH_FIND, &found);
	if (!found)
		{
		/* not found */
		LWLockRelease(lock);
		return false;
		}

	/* found */
	Assert(NULL != entry);
	check_entry_key(tid, entry->key);

	cur_size = strlen(entry->info);
	app_size = strlen(info);
	new_size = cur_size + app_size;
	if (new_size >= MAX_2PC_INFO_SIZE)
		{
		/* overflow */
		LWLockRelease(lock);
		elog(LOG, "[%s] %s new size(%d) overflow(%d)",
			__FUNCTION__, tid, new_size, MAX_2PC_INFO_SIZE);
		*overflow = true;
		return false;
		}

	memcpy(entry->info + cur_size, info, app_size + 1);

	Assert(strlen(entry->info) < MAX_2PC_INFO_SIZE);

	LWLockRelease(lock);

	if (enable_2pc_entry_trace)
			{
		elog(LOG, "[%s] %s is found in hash table", __FUNCTION__, tid);
			}

	return true;
		}

/*
 * remove_2pc_info: remove 2pc info from hash table
 * return true: remove success
 * return false: remove failed
 */
bool remove_2pc_info(const char *tid)
		{
	bool found = false;
	Cache2pcInfo *entry = NULL;
	uint32 hashvalue = Record2pcCacheHashCode(tid);
	LWLock *lock = Cache2pcPartitionLock(hashvalue);

	Assert(NULL != record_2pc_cache);
	Assert(NULL != tid);

	LWLockAcquire(lock, LW_EXCLUSIVE);

	entry = (Cache2pcInfo *)hash_search_with_hash_value(record_2pc_cache,
		tid, hashvalue, HASH_REMOVE, &found);

	LWLockRelease(lock);

		if (!found)
		{
		/* not found */
		Assert(NULL == entry);
		return false;
		}

	/* found */
		Assert (NULL != entry);

	if (enable_2pc_entry_trace)
		{
		elog(LOG, "[%s] %s is removed from hash table, entry: %p",
			__FUNCTION__, tid, entry);
	}

	return true;
}

/*
 * get_2pc_info: get 2pc info from hash table
 * return true: get success
 * return false: get failed
 */
bool get_2pc_info(const char *tid, char *info)
{
    bool found = false;
	Cache2pcInfo *entry = NULL;
	uint32 hashvalue = Record2pcCacheHashCode(tid);
	LWLock *lock = Cache2pcPartitionLock(hashvalue);

	Assert(NULL != record_2pc_cache);
	Assert(NULL != tid);
	Assert(NULL != info);

	LWLockAcquire(lock, LW_SHARED);

	entry = (Cache2pcInfo *)hash_search_with_hash_value(record_2pc_cache,
		tid, hashvalue, HASH_FIND, &found);
	if (!found)
        {
		/* not found */
		LWLockRelease(lock);
		Assert(NULL == entry);
		return false;
    }

	/* found */
	Assert(NULL != entry);
	check_entry_key(tid, entry->key);

	Assert(strlen(entry->info) < MAX_2PC_INFO_SIZE);
	memcpy(info, entry->info, strlen(entry->info) + 1);

	LWLockRelease(lock);
	return true;
}

/*
 * save_and_remove_2pc_info: save 2pc info from hash table to disk file, 
 * then remove it
 * return true: save and remove success
 * return false: save and remove failed
 */
bool save_and_remove_2pc_info(const char *tid)
{
	bool found = false;
	Cache2pcInfo *entry = NULL;
	File fd = -1;
    int ret = 0;
    int size = 0;
    char path[MAXPGPATH];
	uint32 hashvalue = Record2pcCacheHashCode(tid);
	LWLock *lock = Cache2pcPartitionLock(hashvalue);

	Assert(NULL != record_2pc_cache);
	Assert(NULL != tid);
        
	memset(path, 0, MAXPGPATH);
	GET_2PC_FILE_PATH(path, tid);

	LWLockAcquire(lock, LW_EXCLUSIVE);

	/* get 2pc info */
	entry = (Cache2pcInfo *)hash_search_with_hash_value(record_2pc_cache,
		tid, hashvalue, HASH_FIND, &found);
	if (!found)
	{
		/* not found */
		LWLockRelease(lock);
		Assert(NULL == entry);
		return false;
	}

	/* found */
	Assert(NULL != entry);
	check_entry_key(tid, entry->key);

	Assert(strlen(entry->info) < MAX_2PC_INFO_SIZE);

	if (0 == access(path, F_OK))
	{
		/* file exist */
		if (enable_2pc_file_check)
		{
			elog(LOG, "[%s] found file %s", __FUNCTION__, path);
		}

		/* remove file */
		if (0 != unlink(path))
		{
			elog(WARNING, "[%s] could not unlink file %s, errMsg: %s",
				__FUNCTION__, path, strerror(errno));
		}
		else
		{
			elog(LOG, "[%s] unlink file %s", __FUNCTION__, path);
		}
	}

	/* save to file */
	fd = open(path, O_RDWR | O_CREAT | O_EXCL, S_IRUSR | S_IWUSR);
	if (fd < 0)
	{
		LWLockRelease(lock);
		elog(ERROR, "[%s] could not create file %s, errMsg: %s",
			__FUNCTION__, path, strerror(errno));
	}

	size = strlen(entry->info);
	ret = write(fd, entry->info, size);
	if(ret != size)
	{
		LWLockRelease(lock);
		close(fd);
		elog(ERROR, "[%s] could not write file %s, errMsg: %s, "
			"ret: %d, size: %d, info: %s",
			__FUNCTION__, path, strerror(errno), ret, size, entry->info);
	}
	close(fd);

	/* remove 2pc info */
	entry = (Cache2pcInfo *)hash_search_with_hash_value(record_2pc_cache,
		tid, hashvalue, HASH_REMOVE, &found);

	LWLockRelease(lock);

	Assert(found);
	Assert(NULL != entry);

	if (enable_2pc_entry_trace)
	{
		elog(LOG, "[%s] %s is removed from hash table, entry: %p",
			__FUNCTION__, tid, entry);
	}

	return true;
}

/*
 * Check whether the 2pc file is exist when it is saved in the hash table.
 */
void check_2pc_file(const char *tid, const char *info, const char *func)
{
	if (enable_2pc_file_check)
	{
		int size = 0;
		struct stat filestate;
		char path[MAXPGPATH];
		int ret = 0;
		File fd = -1;
		char result[MAX_2PC_INFO_SIZE];

		Assert(NULL != tid);
		Assert(NULL != info);
		Assert(NULL != func);

		GET_2PC_FILE_PATH(path, tid);
		if (0 != access(path, F_OK))
		{
			return;
		}

		elog(LOG, "[check_2pc_file][%s] found file %s", func, path);

		if(stat(path, &filestate) == -1)
		{
			elog(WARNING, "[check_2pc_file][%s] could not stat file %s, info: %s",
				func, path, info);
			return;
		}

		size = filestate.st_size;

		if (0 == size)
		{
			elog(WARNING, "[check_2pc_file][%s] file %s is empty, info: %s",
				func, path, info);
			return;
		}

		if (size >= MAX_2PC_INFO_SIZE)
		{
			elog(WARNING, "[check_2pc_file][%s] file %s size(%d) overflow(%d)",
				func, path, size, MAX_2PC_INFO_SIZE);
			return;
		}

		fd = PathNameOpenFile(path, O_RDONLY, S_IRUSR | S_IWUSR);
		if (fd < 0)
		{
			elog(WARNING, "[check_2pc_file][%s] could not open file %s, "
				"errMsg: %s", func, path, strerror(errno));
			return;
		}

		memset(result, 0, size +1);
		ret = FileRead(fd, result, size, WAIT_EVENT_BUFFILE_READ);
		if(ret != size)
		{
			FileClose(fd);
			elog(WARNING, "[check_2pc_file][%s] could not read file %s, "
				"ret: %d, file size: %d", func, path, ret, size);
			return;
		}
		FileClose(fd);

		if (0 != strcmp(result, info))
		{
			elog(LOG, "[check_2pc_file][%s] file %s mismatch, "
				"result: %s, info: %s", func, path, result, info);
		}
	}
}

void record_2pc_redo_remove_gid_xid(TransactionId xid)
{
	int i;
	GlobalTransaction gxact = NULL;
	bool found = false;

	if(!enable_2pc_recovery_info)
	{
		return;
	}

	for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
	{
		gxact = TwoPhaseState->prepXacts[i];

		if (gxact->xid == xid)
		{
			found = true;
			break;
		}
	}

	Assert(RecoveryInProgress());

	if (found)
	{
		remove_2pc_records(gxact->gid, false);
	}
}

void record_2pc_involved_nodes_xid(const char * tid, 
                                  char * startnode, 
                                  GlobalTransactionId startxid, 
                                  char * nodestring, 
                                  GlobalTransactionId xid)
{
	File fd = 0;
	int ret = 0;
	int size = 0;
	StringInfoData content;
	struct stat fst;
	char path[MAXPGPATH];
	char *result = NULL;

#ifdef __TWO_PHASE_TESTS__
	XLogRecPtr xlogrec = 0;
#endif

	if (!enable_2pc_recovery_info)
	{
		return;
	}

	if (enable_distri_print || enable_2pc_entry_trace)
    {
		elog(LOG, "[%s] record %s, startnode: %s, participants: %s",
			__FUNCTION__, tid, startnode, nodestring);
    }

    if (NULL == tid || '\0' == tid[0])
    {
		elog(ERROR, "[%s] gid is empty", __FUNCTION__);
    }

    if (NULL == startnode || '\0' == startnode[0])
    {
		elog(PANIC, "[%s] %s startnode is empty", __FUNCTION__, tid);
    }
    if (NULL == nodestring || '\0' == nodestring[0])
    {
		elog(PANIC, "[%s] %s participants is empty", __FUNCTION__, tid);
    }
    
	initStringInfo(&content);
	appendStringInfo(&content, "startnode:%s\n", startnode);
	appendStringInfo(&content, "startxid:%u\n", startxid);
	appendStringInfo(&content, "nodes:%s\n", nodestring);
	appendStringInfo(&content, "xid:%u\n", xid);
	size = content.len;
	Assert(size == strlen(content.data));

    /* if in_pg_clean, then check whether the file exists */
    if (g_twophase_state.in_pg_clean)
    {
		/* if tid already exists, check content and return */
		if (NULL != record_2pc_cache)
		{
			char info[MAX_2PC_INFO_SIZE];

			Assert(strlen(tid) < MAX_TID_SIZE);

			if (get_2pc_info(tid, info))
			{
				Assert(strlen(info) < MAX_2PC_INFO_SIZE);
				check_2pc_file(tid, info, __FUNCTION__);

				if (strncmp(info, content.data, size) != 0)
				{
					elog(ERROR, "[%s] pg_clean attemp to write %s info conflict, "
						"content: %s, info: %s", __FUNCTION__, tid,
						content.data, info);
				}

				resetStringInfo(&content);
				pfree(content.data);
				return;
			}
		}

		GET_2PC_FILE_PATH(path, tid);

        /* if file already exists, check content and return */
        if (stat(path, &fst) >= 0)
        {
			int file_size = fst.st_size;
			result = (char *)palloc0(file_size + 1);
            
            fd = PathNameOpenFile(path, O_RDONLY, S_IRUSR | S_IWUSR);
            if (fd < 0)
            {   
				elog(ERROR, "[%s] could not open file %s, errMsg: %s",
					__FUNCTION__, path, strerror(errno));
            } 
			ret = FileRead(fd, result, file_size, WAIT_EVENT_BUFFILE_READ);
			if(ret != file_size)
            {
				FileClose(fd);
				elog(ERROR, "[%s] could not read file %s, ret: %d, file_size: %d",
					__FUNCTION__, path, ret, file_size);
            }
            FileClose(fd);

			Assert(NULL != result);

			if (strncmp(result, content.data, size) != 0)
            {
				elog(ERROR, "[%s] pg_clean attemp to write %s info conflict, "
					"content: %s, info: %s",
					__FUNCTION__, tid, content.data, result);
            }

			pfree(result);

            resetStringInfo(&content);
            pfree(content.data);
            return;
        }
    }
    
	if (!RecoveryInProgress())
	{
		XLogBeginInsert();
		XLogRegisterData((char *)tid, strlen(tid) + 1);
		XLogRegisterData((char *)startnode, strlen(startnode) + 1);
		XLogRegisterData((char *)&startxid, sizeof(GlobalTransactionId) + 1);
		XLogRegisterData((char *)nodestring, strlen(nodestring) + 1);
		XLogRegisterData((char *)&xid, sizeof(GlobalTransactionId) + 1);
#ifdef __TWO_PHASE_TESTS__
		xlogrec = 
#endif
		XLogInsert(RM_XLOG_ID, XLOG_CREATE_2PC_FILE);
#ifdef __TWO_PHASE_TESTS__
		if (PART_PREPARE_AFTER_RECORD_2PC == twophase_exception_case && 
			g_twophase_state.is_start_node)
		{
			XLogFlush(xlogrec);
			run_pg_clean = 1;
			complish = true;
			elog(STOP, "[%s] twophase exception: simulate kill start node "
				"after record 2pc file", __FUNCTION__);
		}
#endif
	}

	if (NULL != record_2pc_cache)
			{
		if (size < MAX_2PC_INFO_SIZE)
				{
			Assert(strlen(tid) < MAX_TID_SIZE);

			if (add_2pc_info(tid, content.data))
				{
				check_2pc_file(tid, content.data, __FUNCTION__);

			resetStringInfo(&content);
			pfree(content.data);

			return;
		}

			elog(LOG, "[%s] %s add to cache failed", __FUNCTION__, tid);
		}
		else
		{
			elog(LOG, "[%s] %s info size(%d) overflow(%d)",
				__FUNCTION__, tid, size, MAX_2PC_INFO_SIZE);
	}
	}

	GET_2PC_FILE_PATH(path, tid);

    /*
     * we open 2pc file under the following two different situations:
     * a. if in recovery mode, 
     *  the existed 2pc file can be trucated and reused.
     * b. if not under recovery progress, 
     *  we not allowed the implicit trans gid existed, 
     *  since the xid in startnode should not be truncate if the twophase trans is part commit or part abort.
     */
    if (RecoveryInProgress())
    {
        fd = PathNameOpenFile(path, O_RDWR | O_TRUNC | O_CREAT, S_IRUSR | S_IWUSR);
    }
    else
    {
        fd = PathNameOpenFile(path, O_RDWR | O_CREAT | O_EXCL, S_IRUSR | S_IWUSR);
    }
    if (fd < 0)
    {   
		elog(ERROR, "[%s] could not create file %s, errMsg: %s",
			__FUNCTION__, path, strerror(errno));
        return;
    }

    ret = FileWrite(fd, content.data, size, WAIT_EVENT_BUFFILE_WRITE);
    if(ret != size)
    {
		FileClose(fd);
		elog(ERROR, "[%s] could not write file %s, errMsg: %s, ret: %d, content: %s",
			__FUNCTION__, path, strerror(errno), ret, content.data);
    }
    FileClose(fd);
    
	resetStringInfo(&content);
	pfree(content.data);
}

/* record commit timestamp in 2pc file while twophase trans failed in commit phase in the current node */
void record_2pc_commit_timestamp(const char *tid, GlobalTimestamp commit_timestamp)
{// #lizard forgives
    char path[MAXPGPATH];
	char file_content[FILE_CONTENT_SIZE];
    StringInfoData content;
	File fd = -1;
	int ret = 0;
	int size = 0;
    XLogRecPtr xlogrec = 0;
    
    if (!enable_2pc_recovery_info)
    {
    	return ;
    }

	if (enable_distri_print || enable_2pc_entry_trace)
    {
		elog(LOG, "[%s] %s commit_timestamp: "INT64_FORMAT,
			__FUNCTION__, tid, commit_timestamp);
    }
    Assert(tid[0] != '\0');
    if (InvalidGlobalTimestamp == commit_timestamp && 
        (TWO_PHASE_COMMITTING == g_twophase_state.state || 
        TWO_PHASE_COMMIT_END == g_twophase_state.state))
    {
		elog(ERROR, "[%s] could not commit transaction '%s' on node '%s' "
			"with InvalidGlobalTimestamp", __FUNCTION__, tid, PGXCNodeName);
    }

	if (!RecoveryInProgress())
	{
		XLogBeginInsert();
		XLogRegisterData((char *)tid, strlen(tid) + 1);
		XLogRegisterData((char *)&commit_timestamp, sizeof(GlobalTimestamp) + 1);
		xlogrec = XLogInsert(RM_XLOG_ID, XLOG_RECORD_2PC_TIMESTAMP);
		/* only start node need to flush and sync XLOG_RECORD_2PC_TIMESTAMP */
		if (IS_PGXC_LOCAL_COORDINATOR)
		{
			XLogFlush(xlogrec);
			SyncRepWaitForLSN(xlogrec, false);
		}
	}
        
	initStringInfo(&content);
	appendStringInfo(&content, "global_commit_timestamp:"INT64_FORMAT"\n",
		commit_timestamp);
	size = content.len;
	Assert(size == strlen(content.data));

	if (NULL != record_2pc_cache)
			{
		bool overflow = false;

		Assert(strlen(tid) < MAX_TID_SIZE);

		if (append_2pc_info(tid, content.data, &overflow))
			{
				resetStringInfo(&content);
				pfree(content.data);
				return;
			}

		if (overflow)
				{
			elog(LOG, "[%s] %s is overflow", __FUNCTION__, tid);

			if (save_and_remove_2pc_info(tid))
				{
				elog(LOG, "[%s] %s save to file", __FUNCTION__, tid);
				}
			else
				{
				elog(LOG, "[%s] %s save to file failed", __FUNCTION__, tid);
				}
			}
		else
		{
			elog(LOG, "[%s] %s is not found in hash table", __FUNCTION__, tid);
		}
		}

	GET_2PC_FILE_PATH(path, tid);

	/* the 2pc file exists already */
	fd = PathNameOpenFile(path, O_RDWR | O_APPEND, S_IRUSR | S_IWUSR);
	if (fd < 0)
	{
        if (RecoveryInProgress())
        {
#if 0            
			int i;
			GlobalTransaction gxact = NULL;
            for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
            {
                gxact = TwoPhaseState->prepXacts[i];
                if (NULL == gxact)
                {
                    continue;
                }
                if (0 == strcmp(gxact->gid, tid))
                {
					elog(ERROR, "[%s] could not append timestamp in file %s, "
						"errMsg: %s", __FUNCTION__, path, strerror(errno));
                }
            }
#endif            
			elog(LOG, "[%s] could not open file %s, errMsg: %s",
				__FUNCTION__, path, strerror(errno));
        }
        else
        {
			elog(ERROR, "[%s] could not open file %s, errMsg: %s",
				__FUNCTION__, path, strerror(errno));
        }
        return;
    }

    if (enable_distri_print)
    {
		memset(file_content, 0, FILE_CONTENT_SIZE);
		ret = FileRead(fd, file_content, FILE_CONTENT_SIZE, WAIT_EVENT_BUFFILE_READ);
		elog(LOG, "[%s] before append file: %s, file_content: %s, content.data: %s, "
			"ret: %d", __FUNCTION__, path, file_content, content.data, ret);
    }

	ret = FileWrite(fd, content.data, size, WAIT_EVENT_BUFFILE_WRITE);
    if(ret != size)
    {
		FileClose(fd);
		elog(ERROR, "[%s] could not write file %s, errMsg: %s",
			__FUNCTION__, path, strerror(errno));
    }

    if (enable_distri_print)
    {
		memset(file_content, 0, FILE_CONTENT_SIZE);
		FileSeek(fd, 0, SEEK_SET);
		ret = FileRead(fd, file_content, FILE_CONTENT_SIZE, WAIT_EVENT_BUFFILE_READ);
		elog(LOG, "[%s] after append file: %s, file_content: %s, ret: %d",
			__FUNCTION__, path, file_content, ret);
    }

	FileClose(fd);

    resetStringInfo(&content);
    pfree(content.data);
}

void remove_2pc_records(const char * tid, bool record_in_xlog)
{
    char path[MAXPGPATH];    

    if (!enable_2pc_recovery_info)
    {
    	return ;
    }

	if (enable_distri_print || enable_2pc_entry_trace)
	{
		elog(LOG, "[%s] %s record_in_xlog: %d", __FUNCTION__, tid, record_in_xlog);
	}

    if (!RecoveryInProgress() && record_in_xlog)
    {
		char *type = "remove";
        XLogBeginInsert();
        XLogRegisterData((char *)tid, strlen(tid)+1);
		XLogRegisterData((char *)type, strlen(type) + 1);
        XLogInsert(RM_XLOG_ID, XLOG_CLEAN_2PC_FILE);
    }

	GET_2PC_FILE_PATH(path, tid);

	if (NULL != record_2pc_cache)
	{
		Assert(strlen(tid) < MAX_TID_SIZE);

		if (enable_2pc_entry_key_check)
		{
			char info[MAX_2PC_INFO_SIZE];
			if (get_2pc_info(tid, info))
			{
				Assert(strlen(info) < MAX_2PC_INFO_SIZE);
				check_2pc_file(tid, info, __FUNCTION__);
			}
		}

		/* remove from hash table */
		if (remove_2pc_info(tid))
		{
			if (enable_2pc_file_check)
			{
				if (0 == access(path, F_OK))
			{
					elog(LOG, "[%s] still found file %s", __FUNCTION__, path);
				}
			}
			return;
		}
	}

	/*
		* no need to check file exists.
		* since when it do not exists, unlink won't success.
	*/
    if (0 != unlink(path))
    {
		elog(LOG, "[%s] could not unlink file %s, errMsg: %s",
			__FUNCTION__, path, strerror(errno));
    }
}

void rename_2pc_records(const char *tid, TimestampTz timestamp)
{
	char path[MAXPGPATH];
	char new_path[MAXPGPATH];

	if (!enable_2pc_recovery_info)
	{
		return;
	}

	if (enable_distri_print || enable_2pc_entry_trace)
	{
		elog(LOG, "[%s] %s timestamp: "INT64_FORMAT, __FUNCTION__, tid, timestamp);
	}

	if (0 == timestamp)
	{
		timestamp = GetCurrentTimestamp();
	}

	if (!RecoveryInProgress())
	{
		char *type = "rename";
		XLogBeginInsert();
		XLogRegisterData((char *)tid, strlen(tid) + 1);
		XLogRegisterData((char *)type, strlen(type) + 1);
		XLogRegisterData((char *)&timestamp, sizeof(TimestampTz) + 1);
		XLogInsert(RM_XLOG_ID, XLOG_CLEAN_2PC_FILE);
	}

	if (NULL != record_2pc_cache)
	{
		Assert(strlen(tid) < MAX_TID_SIZE);

		if (save_and_remove_2pc_info(tid))
			{
			elog(LOG, "[%s] %s save to file", __FUNCTION__, tid);
			}
			else
			{
			elog(LOG, "[%s] %s save to file failed", __FUNCTION__, tid);
				}
			}

	GET_2PC_FILE_PATH(path, tid);
	snprintf(new_path, MAXPGPATH, "%s." INT64_FORMAT ".rollback", path, timestamp);

	if (0 != access(path, F_OK))
			{
		if (RecoveryInProgress())
			{
			elog(LOG, "[%s] could not access file %s in recovery mode, errMsg: %s",
				__FUNCTION__, path, strerror(errno));
			}
		else
			{
			elog(WARNING, "[%s] could not access file %s, errMsg: %s",
				__FUNCTION__, path, strerror(errno));
	}

		return;
	}
	if (0 == access(new_path, F_OK))
	{
		if (RecoveryInProgress())
		{
			elog(LOG, "[%s] file %s exist", __FUNCTION__, new_path);
		}
		else
		{
			elog(WARNING, "[%s] file %s exist", __FUNCTION__, new_path);
		}
		if (0 != unlink(new_path))
		{
			elog(WARNING, "[%s] could not unlink file %s, errMsg: %s",
				__FUNCTION__, new_path, strerror(errno));
			return;
		}
	}
	if (0 != link(path, new_path))
	{
		elog(ERROR, "[%s] could not link file %s to %s, errMsg: %s",
			__FUNCTION__, path, new_path, strerror(errno));
	}
	if (0 != unlink(path))
	{
		elog(WARNING, "[%s] could not unlink file %s, errMsg: %s",
			__FUNCTION__, path, strerror(errno));
	}
}

void record_2pc_readonly(const char *gid)
{
    File fd = 0;
    int ret = 0;
    char path[MAXPGPATH];
	char *content = "readonly";
        
    if(!enable_2pc_recovery_info)
    {
    	return ;
    }
        
	if (enable_distri_print || enable_2pc_entry_trace)
    {
		elog(LOG, "[%s] %s is readonly", __FUNCTION__, gid);
    }

    if (!RecoveryInProgress())
    {
        XLogBeginInsert();
        XLogRegisterData((char *)gid, strlen(gid)+1);
        XLogRegisterData((char *)content, strlen(content)+1);
        XLogInsert(RM_XLOG_ID, XLOG_CREATE_2PC_FILE);
    }

	if (NULL != record_2pc_cache)
	{
		Assert(strlen(gid) < MAX_TID_SIZE);
		Assert(strlen(content) < MAX_2PC_INFO_SIZE);

		if (add_2pc_info(gid, content))
			{
			check_2pc_file(gid, content, __FUNCTION__);
			return;
		}

		elog(LOG, "[%s] %s add to cache failed", __FUNCTION__, gid);
	}

	/* the 2pc dir is already created in initdb */
	GET_2PC_FILE_PATH(path, gid);

    /*
     * we open 2pc file under the following two different situations:
     * a. if in recovery mode, 
     *  the existed 2pc file can be trucated and reused.
     * b. if not under recovery progress, 
     *  we not allowed the implicit trans gid existed, 
		*  since the xid in startnode should not be truncate if the 
		*  twophase trans is part commit or part abort.
     */
    if (RecoveryInProgress())
    {
        fd = PathNameOpenFile(path, O_RDWR | O_TRUNC | O_CREAT, S_IRUSR | S_IWUSR);
    }
    else
    {
        fd = PathNameOpenFile(path, O_RDWR | O_CREAT | O_EXCL, S_IRUSR | S_IWUSR);
    }
    if (fd < 0)
    {   
		elog(ERROR, "[%s] could not create file %s, errMsg: %s",
			__FUNCTION__, path, strerror(errno));
        return;
    }

    ret = FileWrite(fd, content, strlen(content), WAIT_EVENT_BUFFILE_WRITE);
    if(ret != strlen(content))
    {
		FileClose(fd);
		elog(ERROR, "[%s] could not write file %s, errMsg: %s, ret: %d, content: %s",
			__FUNCTION__, path, strerror(errno), ret, content);
    }
    FileClose(fd);
}
    
/*
 * Get 2pc info from hash table.
 */
char *get_2pc_info_from_cache(const char *tid)
{
	char *info = NULL;

	if (NULL == record_2pc_cache)
	{
		return NULL;
	}

		Assert(strlen(tid) < MAX_TID_SIZE);

	info = (char *)palloc0(MAX_2PC_INFO_SIZE);
	if (get_2pc_info(tid, info))
	{
		Assert(strlen(info) < MAX_2PC_INFO_SIZE);
		check_2pc_file(tid, info, __FUNCTION__);

			if (enable_2pc_entry_trace)
			{
			elog(LOG, "[%s] %s is found in hash table", __FUNCTION__, tid);
}

		return info;
		}

	pfree(info);

		if (enable_2pc_entry_trace)
		{
		elog(LOG, "[%s] %s is not found in hash table", __FUNCTION__, tid);
	}

	return NULL;
}

/*
 * Get 2pc list from hash table.
 */
char *get_2pc_list_from_cache(int *count)
{
	HASH_SEQ_STATUS seq;
	Cache2pcInfo *entry = NULL;
	char *recordList = NULL;

	Assert(NULL != count);

	if (NULL == record_2pc_cache)
	{
		return NULL;
	}

	hash_seq_init(&seq, record_2pc_cache);
	while ((entry = hash_seq_search(&seq)) != NULL)
	{
		Assert(NULL != entry);
		check_2pc_file(entry->key, entry->info, __FUNCTION__);

		if(NULL == recordList)
		{
			recordList = (char *)palloc0(strlen(entry->key) + 1);
			sprintf(recordList, "%s", entry->key);
		}
		else
		{
			recordList = (char *) repalloc(recordList,
				strlen(entry->key) + strlen(recordList) + 2);
			sprintf(recordList, "%s,%s", recordList, entry->key);
		}

		if (++(*count) >= MAX_OUTPUT_FILE)
		{
			break;
		}
	}

	return recordList;
}

/*
 * Initialize 2pc info cache using shared memory hash table.
 */
void
Record2pcCacheInit(void)
{
	HASHCTL info;
	int flags = 0;

	if (!enable_2pc_file_cache)
	{
		record_2pc_cache = NULL;
		return;
	}

	info.keysize = MAX_TID_SIZE;
	info.entrysize = record_2pc_entry_size;
	info.num_partitions = NUM_CACHE_2PC_PARTITIONS;

	flags = HASH_ELEM | HASH_PARTITION;

	record_2pc_cache = ShmemInitHash("Record 2pc cache",
		record_2pc_cache_size, record_2pc_cache_size,
		&info, flags);
}

/*
 * Return 2pc info cache size.
 */
Size
Record2pcCacheSize(void)
{
	Size cache_size = 0;
	if (enable_2pc_file_cache)
	{
		cache_size = hash_estimate_size(record_2pc_cache_size, record_2pc_entry_size);
	}
	return cache_size;
}

/*
 * Record2pcCacheHashCode
 *             Compute the hash code associated with a tid
 *
 * This must be passed to the lookup/insert/delete routines along with the
 * tag.  We do it like this because the callers need to know the hash code
 * in order to determine which buffer partition to lock, and we don't want
 * to do the hash computation twice.
 */
uint32
Record2pcCacheHashCode(const char *tid)
{
	Assert(NULL != record_2pc_cache);
	Assert(NULL != tid);
	return get_hash_value(record_2pc_cache, tid);
}

#endif
